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/24 13:32:33 UTC
[GitHub] [iceberg] openinx opened a new pull request #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
openinx opened a new pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818
@rdblue I've continued the work from here https://github.com/apache/iceberg/pull/1802 , and implemented a flink's `RowDataTaskWriter` to accept both insert rows and equality deletions.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530152845
##########
File path: core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
##########
@@ -51,39 +60,149 @@ protected BaseTaskWriter(PartitionSpec spec, FileFormat format, FileAppenderFact
this.targetFileSize = targetFileSize;
}
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected FileAppenderFactory<T> appenderFactory() {
+ return appenderFactory;
+ }
+
@Override
public void abort() throws IOException {
close();
// clean up files created by this writer
- Tasks.foreach(completedFiles)
+ Tasks.foreach(Iterables.concat(completedFiles, completedDeletes))
.throwFailureWhenFinished()
.noRetry()
.run(file -> io.deleteFile(file.path().toString()));
}
@Override
- public DataFile[] complete() throws IOException {
+ public WriterResult complete() throws IOException {
close();
- return completedFiles.toArray(new DataFile[0]);
+ return WriterResult.builder()
+ .addDataFiles(completedFiles)
+ .addDeleteFiles(completedDeletes)
+ .build();
+ }
+
+ protected abstract class BaseDeltaWriter implements Closeable {
+ private final RollingFileWriter dataWriter;
+
+ private final boolean enableEqDelete;
+ private RollingEqDeleteWriter eqDeleteWriter = null;
+ private SortedPosDeleteWriter<T> posDeleteWriter = null;
+ private StructLikeMap<FilePos> insertedRowMap = null;
+
+ public BaseDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds, Schema schema) {
+ this.dataWriter = new RollingFileWriter(partition);
+
+ this.enableEqDelete = equalityFieldIds != null && !equalityFieldIds.isEmpty();
+ if (enableEqDelete) {
Review comment:
> Why use a delta writer if eq deletes are disabled?
Because I only want to expose the `BaseDeltaWriter` to compute engines, I planed to make the `BaseRollingWriter` & `RollingFileWriter` & `RollingEqDeleteWriter` to be `private`. To implement the compute-engine specific `TaskWriter`, the only thing we need to do is implementing the `asKey` and `asCopiedKey` methods and customizing the policy to dispatch records to `DeltaWriter`.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530008671
##########
File path: core/src/main/java/org/apache/iceberg/io/DeleteUtil.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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 org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class DeleteUtil {
+
+ private DeleteUtil() {
+ }
+
+ public static Schema pathPosSchema() {
+ return new Schema(
+ MetadataColumns.DELETE_FILE_PATH,
+ MetadataColumns.DELETE_FILE_POS);
+ }
+
+ public static Schema pathPosSchema(Schema rowSchema) {
Review comment:
This one doesn't need to be public.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530019400
##########
File path: core/src/main/java/org/apache/iceberg/util/StructLikeMap.java
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.util;
+
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Types;
+
+public class StructLikeMap<T> extends AbstractMap<StructLike, T> implements Map<StructLike, T> {
+
+ public static <T> StructLikeMap<T> create(Types.StructType type) {
+ return new StructLikeMap<>(type);
+ }
+
+ private final Types.StructType type;
+ private final Map<StructLikeWrapper, T> wrapperMap;
+ private final ThreadLocal<StructLikeWrapper> wrappers;
+
+ private StructLikeMap(Types.StructType type) {
+ this.type = type;
+ this.wrapperMap = Maps.newHashMap();
+ this.wrappers = ThreadLocal.withInitial(() -> StructLikeWrapper.forType(type));
+ }
+
+ @Override
+ public int size() {
+ return wrapperMap.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return wrapperMap.isEmpty();
+ }
+
+ @Override
+ public boolean containsKey(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ boolean result = wrapperMap.containsKey(wrapper.set((StructLike) key));
+ wrapper.set(null);
+ return result;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean containsValue(Object value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public T get(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ T value = wrapperMap.get(wrapper.set((StructLike) key));
+ wrapper.set(null);
+ return value;
+ }
+ return null;
+ }
+
+ @Override
+ public T put(StructLike key, T value) {
+ return wrapperMap.put(StructLikeWrapper.forType(type).set(key), value);
+ }
+
+ @Override
+ public T remove(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ T value = wrapperMap.remove(wrapper.set((StructLike) key));
+ wrapper.set(null); // don't hold a reference to the value.
+ return value;
+ }
+ return null;
+ }
+
+ @Override
+ public void putAll(Map<? extends StructLike, ? extends T> keyValues) {
+ if (keyValues != null && !keyValues.isEmpty()) {
+ for (Map.Entry<? extends StructLike, ? extends T> pair : keyValues.entrySet()) {
+ wrapperMap.put(StructLikeWrapper.forType(type).set(pair.getKey()), pair.getValue());
+ }
+ }
+ }
+
+ @Override
+ public void clear() {
+ wrapperMap.clear();
+ }
+
+ @Override
+ public Set<StructLike> keySet() {
+ return wrapperMap.keySet().stream().map(StructLikeWrapper::get).collect(Collectors.toSet());
+ }
+
+ @Override
+ public Collection<T> values() {
+ return wrapperMap.values();
+ }
+
+ @Override
+ public Set<Entry<StructLike, T>> entrySet() {
+ throw new UnsupportedOperationException();
Review comment:
I think this should be implemented.
It isn't too difficult to implement `Map.Entry` with a `getKey` method that calls `StructLikeWrapper::get`. This method is commonly called on maps. It's even called from `putAll` above in this class.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530031943
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
+
+ default:
+ throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
+ }
+ }
+
+ @Override
+ public void close() {
+ Tasks.foreach(deltaWriterMap.values())
+ .throwFailureWhenFinished()
+ .noRetry()
+ .run(deltaWriter -> {
+ try {
+ deltaWriter.close();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ }
+
+ private class RowDataDeltaWriter extends BaseDeltaWriter {
+
+ private RowDataDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds) {
+ super(partition, equalityFieldIds, schema);
+ }
+
+ @Override
+ protected StructLike asKey(RowData row) {
+ return rowDataWrapper.wrap(row);
Review comment:
I think this is a bug. It doesn't extract the key fields, it just wraps the row as a `StructLike`. It should extract the equality fields to produce a key, probably using `StructProjection`.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530033227
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
Review comment:
I'm not familiar with `UPDATE_AFTER` or `UPDATE_BEFORE`. Can you help me understand what's going on 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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530022585
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkFileAppenderFactory.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.apache.avro.io.DatumWriter;
+import org.apache.flink.table.data.RowData;
+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.DeleteUtil;
+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;
+
+public class FlinkFileAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
+ private final Schema schema;
+ private final RowType flinkSchema;
+ private final Map<String, String> props;
+ private final PartitionSpec spec;
+ private final List<Integer> equalityFieldIds;
+
+ private final boolean posDeleteWithRow;
+
+ public FlinkFileAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props,
+ PartitionSpec spec, List<Integer> equalityFieldIds) {
+ this.schema = schema;
+ this.flinkSchema = flinkSchema;
+ this.props = props;
+ this.spec = spec;
+ this.equalityFieldIds = equalityFieldIds;
+
+ // TODO Set this switch to true if needed.
+ this.posDeleteWithRow = false;
+ }
+
+ @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(flinkSchema))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(schema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkSchema, msgType))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .rowSchema(schema)
+ .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);
+ Schema rowSchema = posDeleteWithRow ? schema : null;
+ try {
+ switch (format) {
+ case AVRO:
+ Function<org.apache.avro.Schema, DatumWriter<?>> writeFunc = null;
+ if (rowSchema != null) {
+ writeFunc = ignore -> new FlinkAvroWriter(FlinkSchemaUtil.convert(rowSchema));
+ }
+
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(writeFunc)
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(rowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ RowType flinkParquetRowType = FlinkSchemaUtil.convert(DeleteUtil.posDeleteSchema(rowSchema));
+
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkParquetRowType, msgType))
Review comment:
I don't think this will work because the writer that is returned will be wrapped by a `PositionDeleteStructWriter`. That would duplicate the position delete struct because this is going to produce a writer for it as well. That's why the schema passed here is a bug, like I noted above.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530026762
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
##########
@@ -62,7 +62,7 @@ public void open() {
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
// close all open files and emit files to downstream committer operator
- for (DataFile dataFile : writer.complete()) {
+ for (DataFile dataFile : writer.complete().dataFiles()) {
Review comment:
Should this also check that there are no delete files?
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530165685
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Comparators;
+
+public class SortedPosDeleteWriter<T> implements Closeable {
+ private static final int RECORDS_NUM_THRESHOLD = 1000_000;
+
+ private final Map<CharSequence, List<Long>> posDeletes = Maps.newHashMap();
+ private final List<DeleteFile> completedFiles = Lists.newArrayList();
+
+ private final FileAppenderFactory<T> appenderFactory;
+ private final OutputFileFactory fileFactory;
+ private final FileFormat format;
+ private final PartitionKey partitionKey;
+
+ private int records = 0;
+
+ public SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileFormat format,
+ PartitionKey partitionKey) {
+ this.appenderFactory = appenderFactory;
+ this.fileFactory = fileFactory;
+ this.format = format;
+ this.partitionKey = partitionKey;
+ }
+
+ public void delete(CharSequence path, long pos) {
+ delete(path, pos, null);
+ }
+
+ public void delete(CharSequence path, long pos, T row) {
+ // TODO support non-null row in future.
+ Preconditions.checkArgument(row == null, "Does not support non-null row in pos-delete now.");
+
+ posDeletes.compute(path, (k, v) -> {
+ if (v == null) {
+ return Lists.newArrayList(pos);
+ } else {
+ v.add(pos);
+ return v;
+ }
+ });
+
+ records += 1;
+ if (records >= RECORDS_NUM_THRESHOLD) {
+ flush();
+ records = 0;
+ }
+ }
+
+ public List<DeleteFile> complete() {
+ flush();
Review comment:
Yeah, make sense ! It's better to use `close` in this `complete` method.
----------------------------------------------------------------
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 closed pull request #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx closed pull request #1818:
URL: 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] openinx commented on a change in pull request #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530148868
##########
File path: core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
##########
@@ -51,39 +60,149 @@ protected BaseTaskWriter(PartitionSpec spec, FileFormat format, FileAppenderFact
this.targetFileSize = targetFileSize;
}
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected FileAppenderFactory<T> appenderFactory() {
+ return appenderFactory;
+ }
+
@Override
public void abort() throws IOException {
close();
// clean up files created by this writer
- Tasks.foreach(completedFiles)
+ Tasks.foreach(Iterables.concat(completedFiles, completedDeletes))
.throwFailureWhenFinished()
.noRetry()
.run(file -> io.deleteFile(file.path().toString()));
}
@Override
- public DataFile[] complete() throws IOException {
+ public WriterResult complete() throws IOException {
close();
- return completedFiles.toArray(new DataFile[0]);
+ return WriterResult.builder()
+ .addDataFiles(completedFiles)
+ .addDeleteFiles(completedDeletes)
+ .build();
+ }
+
+ protected abstract class BaseDeltaWriter implements Closeable {
+ private final RollingFileWriter dataWriter;
+
+ private final boolean enableEqDelete;
+ private RollingEqDeleteWriter eqDeleteWriter = null;
+ private SortedPosDeleteWriter<T> posDeleteWriter = null;
+ private StructLikeMap<FilePos> insertedRowMap = null;
Review comment:
I'm OK about `map<StructLike, FilePos>`
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530015295
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Comparators;
+
+public class SortedPosDeleteWriter<T> implements Closeable {
+ private static final int RECORDS_NUM_THRESHOLD = 1000_000;
+
+ private final Map<CharSequence, List<Long>> posDeletes = Maps.newHashMap();
+ private final List<DeleteFile> completedFiles = Lists.newArrayList();
+
+ private final FileAppenderFactory<T> appenderFactory;
+ private final OutputFileFactory fileFactory;
+ private final FileFormat format;
+ private final PartitionKey partitionKey;
+
+ private int records = 0;
+
+ public SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileFormat format,
+ PartitionKey partitionKey) {
+ this.appenderFactory = appenderFactory;
+ this.fileFactory = fileFactory;
+ this.format = format;
+ this.partitionKey = partitionKey;
+ }
+
+ public void delete(CharSequence path, long pos) {
+ delete(path, pos, null);
+ }
+
+ public void delete(CharSequence path, long pos, T row) {
+ // TODO support non-null row in future.
+ Preconditions.checkArgument(row == null, "Does not support non-null row in pos-delete now.");
+
+ posDeletes.compute(path, (k, v) -> {
+ if (v == null) {
+ return Lists.newArrayList(pos);
+ } else {
+ v.add(pos);
+ return v;
+ }
+ });
+
+ records += 1;
+ if (records >= RECORDS_NUM_THRESHOLD) {
+ flush();
+ records = 0;
+ }
+ }
+
+ public List<DeleteFile> complete() {
+ flush();
+
+ return completedFiles;
+ }
+
+
+ @Override
+ public void close() throws IOException {
+ flush();
+ }
+
+ private void flush() {
+ if (posDeletes.isEmpty()) {
+ return;
+ }
+
+ EncryptedOutputFile outputFile;
+ if (partitionKey == null) {
+ outputFile = fileFactory.newOutputFile();
+ } else {
+ outputFile = fileFactory.newOutputFile(partitionKey);
+ }
+
+ PositionDeleteWriter<T> writer = appenderFactory.newPosDeleteWriter(outputFile, format, partitionKey);
+ try (PositionDeleteWriter<T> closeableWriter = writer) {
+ CharSequence[] paths = posDeletes.keySet().toArray(new CharSequence[0]);
+ Arrays.sort(paths, Comparators.charSequences());
+
+ for (CharSequence path : paths) {
+ List<Long> positions = posDeletes.get(path);
+ Collections.sort(positions);
+
+ for (Long position : positions) {
+ closeableWriter.delete(path, position);
+ }
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
Review comment:
I always like to include whatever context is available. Here, it may be helpful to know which partition writer failed. What about using `throw new UncheckedIOException("Failed to write position delete file for partition " + partitionKey.toString(), e)`
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Comparators;
+
+public class SortedPosDeleteWriter<T> implements Closeable {
+ private static final int RECORDS_NUM_THRESHOLD = 1000_000;
+
+ private final Map<CharSequence, List<Long>> posDeletes = Maps.newHashMap();
+ private final List<DeleteFile> completedFiles = Lists.newArrayList();
+
+ private final FileAppenderFactory<T> appenderFactory;
+ private final OutputFileFactory fileFactory;
+ private final FileFormat format;
+ private final PartitionKey partitionKey;
+
+ private int records = 0;
+
+ public SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileFormat format,
+ PartitionKey partitionKey) {
+ this.appenderFactory = appenderFactory;
+ this.fileFactory = fileFactory;
+ this.format = format;
+ this.partitionKey = partitionKey;
+ }
+
+ public void delete(CharSequence path, long pos) {
+ delete(path, pos, null);
+ }
+
+ public void delete(CharSequence path, long pos, T row) {
+ // TODO support non-null row in future.
+ Preconditions.checkArgument(row == null, "Does not support non-null row in pos-delete now.");
+
+ posDeletes.compute(path, (k, v) -> {
+ if (v == null) {
+ return Lists.newArrayList(pos);
+ } else {
+ v.add(pos);
+ return v;
+ }
+ });
+
+ records += 1;
+ if (records >= RECORDS_NUM_THRESHOLD) {
+ flush();
+ records = 0;
+ }
+ }
+
+ public List<DeleteFile> complete() {
+ flush();
+
+ return completedFiles;
+ }
+
+
+ @Override
+ public void close() throws IOException {
+ flush();
+ }
+
+ private void flush() {
+ if (posDeletes.isEmpty()) {
+ return;
+ }
+
+ EncryptedOutputFile outputFile;
+ if (partitionKey == null) {
+ outputFile = fileFactory.newOutputFile();
+ } else {
+ outputFile = fileFactory.newOutputFile(partitionKey);
+ }
+
+ PositionDeleteWriter<T> writer = appenderFactory.newPosDeleteWriter(outputFile, format, partitionKey);
+ try (PositionDeleteWriter<T> closeableWriter = writer) {
+ CharSequence[] paths = posDeletes.keySet().toArray(new CharSequence[0]);
+ Arrays.sort(paths, Comparators.charSequences());
+
+ for (CharSequence path : paths) {
+ List<Long> positions = posDeletes.get(path);
+ Collections.sort(positions);
+
+ for (Long position : positions) {
+ closeableWriter.delete(path, position);
+ }
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
Review comment:
I always like to include whatever context is available. Here, it may be helpful to know which partition writer failed. What about using `throw new UncheckedIOException("Failed to write position delete file for partition " + partitionKey, e)`
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530171649
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -95,4 +102,24 @@ 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 outputFile, FileFormat format,
+ StructLike partition) {
+ throw new UnsupportedOperationException("Cannot create equality-delete writer for generic record now.");
Review comment:
I expected to provide the implemented methods with complete coverage unit tests. But for the purpose providing the PoC solution as soon as possible, I did't have the time to write those tests so left them unsupported now.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530020319
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Comparators;
+
+public class SortedPosDeleteWriter<T> implements Closeable {
+ private static final int RECORDS_NUM_THRESHOLD = 1000_000;
+
+ private final Map<CharSequence, List<Long>> posDeletes = Maps.newHashMap();
Review comment:
Shouldn't this use `StructLikeMap`?
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530034196
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
+
+ default:
+ throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
+ }
+ }
+
+ @Override
+ public void close() {
+ Tasks.foreach(deltaWriterMap.values())
+ .throwFailureWhenFinished()
+ .noRetry()
+ .run(deltaWriter -> {
Review comment:
You can supply a checked exception class that will be thrown. That way, you can move the try/catch outside of the `run` block:
```java
@Override
public void close() {
try {
Tasks.foreach(deltaWriterMap.values())
.throwFailureWhenFinished()
.noRetry()
.run(RowDataDeltaWriter::close);
} catch (IOException e) {
throw new UncheckedIOException("Failed to close writers", e);
}
}
```
And you can also add a callback that receives the exception for each failure with `onFailure` if you want to log the exceptions individually.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530007451
##########
File path: core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
##########
@@ -51,39 +60,149 @@ protected BaseTaskWriter(PartitionSpec spec, FileFormat format, FileAppenderFact
this.targetFileSize = targetFileSize;
}
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected FileAppenderFactory<T> appenderFactory() {
+ return appenderFactory;
+ }
+
@Override
public void abort() throws IOException {
close();
// clean up files created by this writer
- Tasks.foreach(completedFiles)
+ Tasks.foreach(Iterables.concat(completedFiles, completedDeletes))
.throwFailureWhenFinished()
.noRetry()
.run(file -> io.deleteFile(file.path().toString()));
}
@Override
- public DataFile[] complete() throws IOException {
+ public WriterResult complete() throws IOException {
close();
- return completedFiles.toArray(new DataFile[0]);
+ return WriterResult.builder()
+ .addDataFiles(completedFiles)
+ .addDeleteFiles(completedDeletes)
+ .build();
+ }
+
+ protected abstract class BaseDeltaWriter implements Closeable {
+ private final RollingFileWriter dataWriter;
+
+ private final boolean enableEqDelete;
+ private RollingEqDeleteWriter eqDeleteWriter = null;
+ private SortedPosDeleteWriter<T> posDeleteWriter = null;
+ private StructLikeMap<FilePos> insertedRowMap = null;
+
+ public BaseDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds, Schema schema) {
Review comment:
The list `equalityFieldIds` is only used in this constructor and it is used to create a projection of the schema that is passed in. I think it would be better to pass the delete schema or null in, so that we don't need each writer to create a new projection of the row 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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r529998358
##########
File path: core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
##########
@@ -51,39 +60,149 @@ protected BaseTaskWriter(PartitionSpec spec, FileFormat format, FileAppenderFact
this.targetFileSize = targetFileSize;
}
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected FileAppenderFactory<T> appenderFactory() {
+ return appenderFactory;
+ }
+
@Override
public void abort() throws IOException {
close();
// clean up files created by this writer
- Tasks.foreach(completedFiles)
+ Tasks.foreach(Iterables.concat(completedFiles, completedDeletes))
.throwFailureWhenFinished()
.noRetry()
.run(file -> io.deleteFile(file.path().toString()));
}
@Override
- public DataFile[] complete() throws IOException {
+ public WriterResult complete() throws IOException {
close();
- return completedFiles.toArray(new DataFile[0]);
+ return WriterResult.builder()
+ .addDataFiles(completedFiles)
+ .addDeleteFiles(completedDeletes)
+ .build();
+ }
+
+ protected abstract class BaseDeltaWriter implements Closeable {
+ private final RollingFileWriter dataWriter;
+
+ private final boolean enableEqDelete;
+ private RollingEqDeleteWriter eqDeleteWriter = null;
+ private SortedPosDeleteWriter<T> posDeleteWriter = null;
+ private StructLikeMap<FilePos> insertedRowMap = null;
Review comment:
Minor: Do we need to use `StructLikeMap` in declarations or can we use `Map<StructLike, FilePos>`? The latter seems easier to read, even if the map implementation is `StructLikeMap`.
I think the risk is someone accidentally replacing the implementation with a hash map, but we have that problem in places where we use `Maps.newTreeMap()` 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] openinx commented on a change in pull request #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r532571429
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
Review comment:
> Is it that it will be associated with UPDATE_AFTER so the two are combined to form a replace operation?
Yes, that's right.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530589968
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
Review comment:
What distinguishes `UPDATE_BEFORE` from `DELETE`? Is it that it will be associated with `UPDATE_AFTER` so the two are combined to form a replace operation?
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530021516
##########
File path: data/src/test/java/org/apache/iceberg/TestSplitScan.java
##########
@@ -63,7 +63,7 @@
@Parameterized.Parameters(name = "format = {0}")
public static Object[] parameters() {
- return new Object[] { "parquet", "avro" };
+ return new Object[] {"parquet", "avro"};
Review comment:
Nit: this line doesn't need to change. Can you revert it to avoid commit conflicts?
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530164659
##########
File path: core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
##########
@@ -51,39 +60,149 @@ protected BaseTaskWriter(PartitionSpec spec, FileFormat format, FileAppenderFact
this.targetFileSize = targetFileSize;
}
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected FileAppenderFactory<T> appenderFactory() {
+ return appenderFactory;
+ }
+
@Override
public void abort() throws IOException {
close();
// clean up files created by this writer
- Tasks.foreach(completedFiles)
+ Tasks.foreach(Iterables.concat(completedFiles, completedDeletes))
.throwFailureWhenFinished()
.noRetry()
.run(file -> io.deleteFile(file.path().toString()));
}
@Override
- public DataFile[] complete() throws IOException {
+ public WriterResult complete() throws IOException {
close();
- return completedFiles.toArray(new DataFile[0]);
+ return WriterResult.builder()
+ .addDataFiles(completedFiles)
+ .addDeleteFiles(completedDeletes)
+ .build();
+ }
+
+ protected abstract class BaseDeltaWriter implements Closeable {
+ private final RollingFileWriter dataWriter;
+
+ private final boolean enableEqDelete;
+ private RollingEqDeleteWriter eqDeleteWriter = null;
+ private SortedPosDeleteWriter<T> posDeleteWriter = null;
+ private StructLikeMap<FilePos> insertedRowMap = null;
+
+ public BaseDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds, Schema schema) {
+ this.dataWriter = new RollingFileWriter(partition);
+
+ this.enableEqDelete = equalityFieldIds != null && !equalityFieldIds.isEmpty();
+ if (enableEqDelete) {
+ this.eqDeleteWriter = new RollingEqDeleteWriter(partition);
+ this.posDeleteWriter = new SortedPosDeleteWriter<>(appenderFactory, fileFactory, format, partition);
+
+ Schema deleteSchema = TypeUtil.select(schema, Sets.newHashSet(equalityFieldIds));
+ this.insertedRowMap = StructLikeMap.create(deleteSchema.asStruct());
+ }
+ }
+
+ protected abstract StructLike asKey(T row);
+
+ protected abstract StructLike asCopiedKey(T row);
+
+ public void write(T row) throws IOException {
+ if (enableEqDelete) {
+ FilePos filePos = FilePos.create(dataWriter.currentPath(), dataWriter.currentRows());
+
+ StructLike copiedKey = asCopiedKey(row);
+ // Adding a pos-delete to replace the old filePos.
+ FilePos previous = insertedRowMap.put(copiedKey, filePos);
+ if (previous != null) {
+ posDeleteWriter.delete(previous.path, previous.rowOffset, null /* TODO set non-nullable row*/);
Review comment:
The straightforward way is adding a `row` field in `FilePos` which will reference to the inserted old row, but that will hold references of all the inserted rows in a checkpoint. If the row is large while equality fields are small, then the idea way should only keep the equality fields & file-pos in the `insertedRowMap`, but if we want to attach `row` when writing pos-delete file then the memory consumption is an issue. I'm considering that maybe we will need an embedded KV lib which could split to disk in future.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530034196
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
+
+ default:
+ throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
+ }
+ }
+
+ @Override
+ public void close() {
+ Tasks.foreach(deltaWriterMap.values())
+ .throwFailureWhenFinished()
+ .noRetry()
+ .run(deltaWriter -> {
Review comment:
You can supply a checked exception class that will be thrown. That way, you can move the try/catch outside of the `run` block:
```java
@Override
public void close() {
try {
Tasks.foreach(deltaWriterMap.values())
.throwFailureWhenFinished()
.noRetry()
.run(RowDataDeltaWriter::close, IOException.class);
} catch (IOException e) {
throw new UncheckedIOException("Failed to close writers", e);
}
}
```
And you can also add a callback that receives the exception for each failure with `onFailure` if you want to log the exceptions individually.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530016819
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Comparators;
+
+public class SortedPosDeleteWriter<T> implements Closeable {
+ private static final int RECORDS_NUM_THRESHOLD = 1000_000;
+
+ private final Map<CharSequence, List<Long>> posDeletes = Maps.newHashMap();
+ private final List<DeleteFile> completedFiles = Lists.newArrayList();
+
+ private final FileAppenderFactory<T> appenderFactory;
+ private final OutputFileFactory fileFactory;
+ private final FileFormat format;
+ private final PartitionKey partitionKey;
+
+ private int records = 0;
+
+ public SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileFormat format,
+ PartitionKey partitionKey) {
+ this.appenderFactory = appenderFactory;
+ this.fileFactory = fileFactory;
+ this.format = format;
+ this.partitionKey = partitionKey;
+ }
+
+ public void delete(CharSequence path, long pos) {
+ delete(path, pos, null);
+ }
+
+ public void delete(CharSequence path, long pos, T row) {
+ // TODO support non-null row in future.
+ Preconditions.checkArgument(row == null, "Does not support non-null row in pos-delete now.");
+
+ posDeletes.compute(path, (k, v) -> {
+ if (v == null) {
+ return Lists.newArrayList(pos);
+ } else {
+ v.add(pos);
+ return v;
+ }
+ });
+
+ records += 1;
+ if (records >= RECORDS_NUM_THRESHOLD) {
+ flush();
+ records = 0;
+ }
+ }
+
+ public List<DeleteFile> complete() {
+ flush();
Review comment:
I would expect this to call `close` rather than `flush`. While `close` just calls `flush` so they are equivalent right now, I think using `close` is better in the long term. If `close` is modified in the future, it is unlikely that someone will go here and make the same change.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530179342
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
Review comment:
For CDC events, such as mysql binlog: `UPDATE test set a=2, b=2 where a=1, b=1`, the [flink-cdc-connectors](https://github.com/ververica/flink-cdc-connectors) will produce two change logs for it:
```
-U (a=1, b=1)
+U (a=2,b=2)
```
The first `RowData` means we will need to delete the old row (1,1), it's also called a `UPDATE_BEFORE` row. The second `RowData` means we will need to insert the new row (2,2), it's also called a `UPDATE_AFTER` 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 commented on a change in pull request #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530184513
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
+
+ default:
+ throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
+ }
+ }
+
+ @Override
+ public void close() {
+ Tasks.foreach(deltaWriterMap.values())
+ .throwFailureWhenFinished()
+ .noRetry()
+ .run(deltaWriter -> {
+ try {
+ deltaWriter.close();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ }
+
+ private class RowDataDeltaWriter extends BaseDeltaWriter {
+
+ private RowDataDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds) {
+ super(partition, equalityFieldIds, schema);
+ }
+
+ @Override
+ protected StructLike asKey(RowData row) {
+ return rowDataWrapper.wrap(row);
Review comment:
The tests work because the `StructLikeMap` has a `StructLikeWrapper` which will only compare the equality fields, even here we provided the full columns. The name `asKey` and `asCopiedKey` are not inappropriate here, `asStructLike` and `asCopiedStructLike` will be better.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530032341
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
Review comment:
You can use `spec.isUnpartitioned()`.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530584524
##########
File path: core/src/main/java/org/apache/iceberg/TableMetadata.java
##########
@@ -64,22 +66,25 @@ public static TableMetadata newTableMetadata(TableOperations ops,
PartitionSpec spec,
String location,
Map<String, String> properties) {
- return newTableMetadata(schema, spec, SortOrder.unsorted(), location, properties, DEFAULT_TABLE_FORMAT_VERSION);
+ int formatVersion = PropertyUtil.propertyAsInt(properties, FORMAT_VERSION, FORMAT_VERSION_DEFAULT);
+ return newTableMetadata(schema, spec, SortOrder.unsorted(), location, properties, formatVersion);
Review comment:
Yes, that's the plan.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530005447
##########
File path: core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
##########
@@ -51,39 +60,149 @@ protected BaseTaskWriter(PartitionSpec spec, FileFormat format, FileAppenderFact
this.targetFileSize = targetFileSize;
}
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected FileAppenderFactory<T> appenderFactory() {
+ return appenderFactory;
+ }
+
@Override
public void abort() throws IOException {
close();
// clean up files created by this writer
- Tasks.foreach(completedFiles)
+ Tasks.foreach(Iterables.concat(completedFiles, completedDeletes))
.throwFailureWhenFinished()
.noRetry()
.run(file -> io.deleteFile(file.path().toString()));
}
@Override
- public DataFile[] complete() throws IOException {
+ public WriterResult complete() throws IOException {
close();
- return completedFiles.toArray(new DataFile[0]);
+ return WriterResult.builder()
+ .addDataFiles(completedFiles)
+ .addDeleteFiles(completedDeletes)
+ .build();
+ }
+
+ protected abstract class BaseDeltaWriter implements Closeable {
+ private final RollingFileWriter dataWriter;
+
+ private final boolean enableEqDelete;
+ private RollingEqDeleteWriter eqDeleteWriter = null;
+ private SortedPosDeleteWriter<T> posDeleteWriter = null;
+ private StructLikeMap<FilePos> insertedRowMap = null;
+
+ public BaseDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds, Schema schema) {
+ this.dataWriter = new RollingFileWriter(partition);
+
+ this.enableEqDelete = equalityFieldIds != null && !equalityFieldIds.isEmpty();
+ if (enableEqDelete) {
+ this.eqDeleteWriter = new RollingEqDeleteWriter(partition);
+ this.posDeleteWriter = new SortedPosDeleteWriter<>(appenderFactory, fileFactory, format, partition);
+
+ Schema deleteSchema = TypeUtil.select(schema, Sets.newHashSet(equalityFieldIds));
+ this.insertedRowMap = StructLikeMap.create(deleteSchema.asStruct());
+ }
+ }
+
+ protected abstract StructLike asKey(T row);
+
+ protected abstract StructLike asCopiedKey(T row);
+
+ public void write(T row) throws IOException {
+ if (enableEqDelete) {
+ FilePos filePos = FilePos.create(dataWriter.currentPath(), dataWriter.currentRows());
+
+ StructLike copiedKey = asCopiedKey(row);
+ // Adding a pos-delete to replace the old filePos.
+ FilePos previous = insertedRowMap.put(copiedKey, filePos);
+ if (previous != null) {
+ posDeleteWriter.delete(previous.path, previous.rowOffset, null /* TODO set non-nullable row*/);
+ }
+ }
+
+ dataWriter.write(row);
+ }
+
+ public void delete(T row) throws IOException {
+ Preconditions.checkState(enableEqDelete, "Could not accept equality deletion.");
+
+ StructLike key = asKey(row);
+ FilePos previous = insertedRowMap.remove(key);
+
+ if (previous != null) {
+ posDeleteWriter.delete(previous.path, previous.rowOffset, null /* TODO set non-nullable row */);
+ }
+
+ eqDeleteWriter.write(row);
+ }
+
+ @Override
+ public void close() throws IOException {
+ // Moving the completed data files into task writer's completedFiles automatically.
+ dataWriter.close();
Review comment:
Minor: `dataWriter` should be set to `null` so that it can be garbage collected and so any further calls to write will fail.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530028470
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java
##########
@@ -562,6 +562,7 @@ public void setColumnStore(ColumnWriteStore columnStore) {
}
public static class PositionDeleteStructWriter<R> extends StructWriter<PositionDelete<R>> {
+
Review comment:
Nit: this file doesn't need to change.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530017205
##########
File path: core/src/main/java/org/apache/iceberg/io/WriterResult.java
##########
@@ -0,0 +1,124 @@
+/*
+ * 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.Serializable;
+import java.util.List;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+public class WriterResult implements Serializable {
+ private static final DataFile[] EMPTY_DATA_FILES = new DataFile[0];
+ private static final DeleteFile[] EMPTY_DELETE_FILES = new DeleteFile[0];
+
+ private DataFile[] dataFiles;
+ private DeleteFile[] deleteFiles;
+
+ private WriterResult(DataFile[] dataFiles, DeleteFile[] deleteFiles) {
+ this.dataFiles = dataFiles;
+ this.deleteFiles = deleteFiles;
+ }
+
+ private WriterResult(List<DataFile> dataFiles, List<DeleteFile> deleteFiles) {
+ this.dataFiles = dataFiles.toArray(new DataFile[0]);
+ this.deleteFiles = deleteFiles.toArray(new DeleteFile[0]);
+ }
+
+ public DataFile[] dataFiles() {
+ return dataFiles;
+ }
+
+ public DeleteFile[] deleteFiles() {
+ return deleteFiles;
+ }
+
+ public static WriterResult create(DataFile dataFile) {
+ return new WriterResult(new DataFile[] {dataFile}, EMPTY_DELETE_FILES);
+ }
+
+ public static WriterResult create(DeleteFile deleteFile) {
+ return new WriterResult(EMPTY_DATA_FILES, new DeleteFile[] {deleteFile});
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static class Builder {
+ private final List<DataFile> dataFiles;
+ private final List<DeleteFile> deleteFiles;
+
+ private Builder() {
+ this.dataFiles = Lists.newArrayList();
+ this.deleteFiles = Lists.newArrayList();
+ }
+
+ public Builder add(WriterResult result) {
+ for (DataFile dataFile : result.dataFiles()) {
+ add(dataFile);
+ }
+ for (DeleteFile deleteFile : result.deleteFiles()) {
+ add(deleteFile);
+ }
Review comment:
Rather than copying the loops, could this call `addDataFiles` and `addDeleteFiles`?
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530021312
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -95,4 +102,24 @@ 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 outputFile, FileFormat format,
+ StructLike partition) {
+ throw new UnsupportedOperationException("Cannot create equality-delete writer for generic record now.");
Review comment:
Why not implement this?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530169794
##########
File path: core/src/main/java/org/apache/iceberg/util/StructLikeMap.java
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.util;
+
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Types;
+
+public class StructLikeMap<T> extends AbstractMap<StructLike, T> implements Map<StructLike, T> {
+
+ public static <T> StructLikeMap<T> create(Types.StructType type) {
+ return new StructLikeMap<>(type);
+ }
+
+ private final Types.StructType type;
+ private final Map<StructLikeWrapper, T> wrapperMap;
+ private final ThreadLocal<StructLikeWrapper> wrappers;
+
+ private StructLikeMap(Types.StructType type) {
+ this.type = type;
+ this.wrapperMap = Maps.newHashMap();
+ this.wrappers = ThreadLocal.withInitial(() -> StructLikeWrapper.forType(type));
+ }
+
+ @Override
+ public int size() {
+ return wrapperMap.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return wrapperMap.isEmpty();
+ }
+
+ @Override
+ public boolean containsKey(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ boolean result = wrapperMap.containsKey(wrapper.set((StructLike) key));
+ wrapper.set(null);
+ return result;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean containsValue(Object value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public T get(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ T value = wrapperMap.get(wrapper.set((StructLike) key));
+ wrapper.set(null);
+ return value;
+ }
+ return null;
+ }
+
+ @Override
+ public T put(StructLike key, T value) {
+ return wrapperMap.put(StructLikeWrapper.forType(type).set(key), value);
+ }
+
+ @Override
+ public T remove(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ T value = wrapperMap.remove(wrapper.set((StructLike) key));
+ wrapper.set(null); // don't hold a reference to the value.
+ return value;
+ }
+ return null;
+ }
+
+ @Override
+ public void putAll(Map<? extends StructLike, ? extends T> keyValues) {
+ if (keyValues != null && !keyValues.isEmpty()) {
+ for (Map.Entry<? extends StructLike, ? extends T> pair : keyValues.entrySet()) {
+ wrapperMap.put(StructLikeWrapper.forType(type).set(pair.getKey()), pair.getValue());
+ }
+ }
+ }
+
+ @Override
+ public void clear() {
+ wrapperMap.clear();
+ }
+
+ @Override
+ public Set<StructLike> keySet() {
+ return wrapperMap.keySet().stream().map(StructLikeWrapper::get).collect(Collectors.toSet());
Review comment:
Make sense. I will provide a full unit test to cover all those map API.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530027845
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkFileAppenderFactory.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.apache.avro.io.DatumWriter;
+import org.apache.flink.table.data.RowData;
+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.DeleteUtil;
+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;
+
+public class FlinkFileAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
+ private final Schema schema;
+ private final RowType flinkSchema;
+ private final Map<String, String> props;
+ private final PartitionSpec spec;
+ private final List<Integer> equalityFieldIds;
+
+ private final boolean posDeleteWithRow;
+
+ public FlinkFileAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props,
+ PartitionSpec spec, List<Integer> equalityFieldIds) {
+ this.schema = schema;
+ this.flinkSchema = flinkSchema;
+ this.props = props;
+ this.spec = spec;
+ this.equalityFieldIds = equalityFieldIds;
+
+ // TODO Set this switch to true if needed.
+ this.posDeleteWithRow = false;
+ }
+
+ @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(flinkSchema))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(schema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkSchema, msgType))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .rowSchema(schema)
+ .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);
+ Schema rowSchema = posDeleteWithRow ? schema : null;
+ try {
+ switch (format) {
+ case AVRO:
+ Function<org.apache.avro.Schema, DatumWriter<?>> writeFunc = null;
+ if (rowSchema != null) {
+ writeFunc = ignore -> new FlinkAvroWriter(FlinkSchemaUtil.convert(rowSchema));
+ }
+
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(writeFunc)
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(rowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ RowType flinkParquetRowType = FlinkSchemaUtil.convert(DeleteUtil.posDeleteSchema(rowSchema));
+
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkParquetRowType, msgType))
Review comment:
Yeah, this needs to be fixed. I'm looking through how Avro actually works right now and it is okay because we're calling `setSchema` a second time from the position writer, which basically discards the original writer with the extra record and rebuilds it with just the row schema.
I'll open a PR to fix 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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r529984488
##########
File path: core/src/main/java/org/apache/iceberg/TableMetadata.java
##########
@@ -64,22 +66,25 @@ public static TableMetadata newTableMetadata(TableOperations ops,
PartitionSpec spec,
String location,
Map<String, String> properties) {
- return newTableMetadata(schema, spec, SortOrder.unsorted(), location, properties, DEFAULT_TABLE_FORMAT_VERSION);
+ int formatVersion = PropertyUtil.propertyAsInt(properties, FORMAT_VERSION, FORMAT_VERSION_DEFAULT);
+ return newTableMetadata(schema, spec, SortOrder.unsorted(), location, properties, formatVersion);
Review comment:
Right now, we don't make it easy to create a v2 table because we don't want people actually using them. That's why all of the tests currently create a table and then update it to v2.
We would also need to remove `FORMAT_VERSION` from properties because we don't intend to keep a table property in sync with the table version in metadata.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530589332
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
+
+ default:
+ throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
+ }
+ }
+
+ @Override
+ public void close() {
+ Tasks.foreach(deltaWriterMap.values())
+ .throwFailureWhenFinished()
+ .noRetry()
+ .run(deltaWriter -> {
+ try {
+ deltaWriter.close();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ }
+
+ private class RowDataDeltaWriter extends BaseDeltaWriter {
+
+ private RowDataDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds) {
+ super(partition, equalityFieldIds, schema);
+ }
+
+ @Override
+ protected StructLike asKey(RowData row) {
+ return rowDataWrapper.wrap(row);
Review comment:
> The tests work because the StructLikeMap has a StructLikeWrapper which will only compare the equality fields
But this happens using the key schema and fields are accessed by position. Wouldn't that fail if the key schema wasn't a prefix of the row 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] openinx commented on a change in pull request #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530167346
##########
File path: core/src/main/java/org/apache/iceberg/io/WriterResult.java
##########
@@ -0,0 +1,124 @@
+/*
+ * 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.Serializable;
+import java.util.List;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+public class WriterResult implements Serializable {
+ private static final DataFile[] EMPTY_DATA_FILES = new DataFile[0];
+ private static final DeleteFile[] EMPTY_DELETE_FILES = new DeleteFile[0];
+
+ private DataFile[] dataFiles;
+ private DeleteFile[] deleteFiles;
+
+ private WriterResult(DataFile[] dataFiles, DeleteFile[] deleteFiles) {
+ this.dataFiles = dataFiles;
+ this.deleteFiles = deleteFiles;
+ }
+
+ private WriterResult(List<DataFile> dataFiles, List<DeleteFile> deleteFiles) {
+ this.dataFiles = dataFiles.toArray(new DataFile[0]);
+ this.deleteFiles = deleteFiles.toArray(new DeleteFile[0]);
+ }
+
+ public DataFile[] dataFiles() {
+ return dataFiles;
+ }
+
+ public DeleteFile[] deleteFiles() {
+ return deleteFiles;
+ }
+
+ public static WriterResult create(DataFile dataFile) {
+ return new WriterResult(new DataFile[] {dataFile}, EMPTY_DELETE_FILES);
+ }
+
+ public static WriterResult create(DeleteFile deleteFile) {
+ return new WriterResult(EMPTY_DATA_FILES, new DeleteFile[] {deleteFile});
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static class Builder {
+ private final List<DataFile> dataFiles;
+ private final List<DeleteFile> deleteFiles;
+
+ private Builder() {
+ this.dataFiles = Lists.newArrayList();
+ this.deleteFiles = Lists.newArrayList();
+ }
+
+ public Builder add(WriterResult result) {
+ for (DataFile dataFile : result.dataFiles()) {
+ add(dataFile);
+ }
+ for (DeleteFile deleteFile : result.deleteFiles()) {
+ add(deleteFile);
+ }
Review comment:
The `dataFiles()` will return an array (I defined it as an array because wanting to avoid the serialization issue for ImmutableList), while the current `addDataFiles` would only accept `Iterable<DataFile>`. I think I can add a `addDataFiles(DataFile... dataFiles)`.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530003061
##########
File path: core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
##########
@@ -51,39 +60,149 @@ protected BaseTaskWriter(PartitionSpec spec, FileFormat format, FileAppenderFact
this.targetFileSize = targetFileSize;
}
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected FileAppenderFactory<T> appenderFactory() {
+ return appenderFactory;
+ }
+
@Override
public void abort() throws IOException {
close();
// clean up files created by this writer
- Tasks.foreach(completedFiles)
+ Tasks.foreach(Iterables.concat(completedFiles, completedDeletes))
.throwFailureWhenFinished()
.noRetry()
.run(file -> io.deleteFile(file.path().toString()));
}
@Override
- public DataFile[] complete() throws IOException {
+ public WriterResult complete() throws IOException {
close();
- return completedFiles.toArray(new DataFile[0]);
+ return WriterResult.builder()
+ .addDataFiles(completedFiles)
+ .addDeleteFiles(completedDeletes)
+ .build();
+ }
+
+ protected abstract class BaseDeltaWriter implements Closeable {
+ private final RollingFileWriter dataWriter;
+
+ private final boolean enableEqDelete;
+ private RollingEqDeleteWriter eqDeleteWriter = null;
+ private SortedPosDeleteWriter<T> posDeleteWriter = null;
+ private StructLikeMap<FilePos> insertedRowMap = null;
+
+ public BaseDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds, Schema schema) {
+ this.dataWriter = new RollingFileWriter(partition);
+
+ this.enableEqDelete = equalityFieldIds != null && !equalityFieldIds.isEmpty();
+ if (enableEqDelete) {
+ this.eqDeleteWriter = new RollingEqDeleteWriter(partition);
+ this.posDeleteWriter = new SortedPosDeleteWriter<>(appenderFactory, fileFactory, format, partition);
+
+ Schema deleteSchema = TypeUtil.select(schema, Sets.newHashSet(equalityFieldIds));
+ this.insertedRowMap = StructLikeMap.create(deleteSchema.asStruct());
+ }
+ }
+
+ protected abstract StructLike asKey(T row);
+
+ protected abstract StructLike asCopiedKey(T row);
+
+ public void write(T row) throws IOException {
+ if (enableEqDelete) {
+ FilePos filePos = FilePos.create(dataWriter.currentPath(), dataWriter.currentRows());
+
+ StructLike copiedKey = asCopiedKey(row);
+ // Adding a pos-delete to replace the old filePos.
+ FilePos previous = insertedRowMap.put(copiedKey, filePos);
+ if (previous != null) {
+ posDeleteWriter.delete(previous.path, previous.rowOffset, null /* TODO set non-nullable row*/);
Review comment:
How would this set the row? Would we need to keep track of it somehow?
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530000642
##########
File path: core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java
##########
@@ -51,39 +60,149 @@ protected BaseTaskWriter(PartitionSpec spec, FileFormat format, FileAppenderFact
this.targetFileSize = targetFileSize;
}
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected FileAppenderFactory<T> appenderFactory() {
+ return appenderFactory;
+ }
+
@Override
public void abort() throws IOException {
close();
// clean up files created by this writer
- Tasks.foreach(completedFiles)
+ Tasks.foreach(Iterables.concat(completedFiles, completedDeletes))
.throwFailureWhenFinished()
.noRetry()
.run(file -> io.deleteFile(file.path().toString()));
}
@Override
- public DataFile[] complete() throws IOException {
+ public WriterResult complete() throws IOException {
close();
- return completedFiles.toArray(new DataFile[0]);
+ return WriterResult.builder()
+ .addDataFiles(completedFiles)
+ .addDeleteFiles(completedDeletes)
+ .build();
+ }
+
+ protected abstract class BaseDeltaWriter implements Closeable {
+ private final RollingFileWriter dataWriter;
+
+ private final boolean enableEqDelete;
+ private RollingEqDeleteWriter eqDeleteWriter = null;
+ private SortedPosDeleteWriter<T> posDeleteWriter = null;
+ private StructLikeMap<FilePos> insertedRowMap = null;
+
+ public BaseDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds, Schema schema) {
+ this.dataWriter = new RollingFileWriter(partition);
+
+ this.enableEqDelete = equalityFieldIds != null && !equalityFieldIds.isEmpty();
+ if (enableEqDelete) {
Review comment:
Why use a delta writer if eq deletes are disabled?
I typically like to use classes that don't need to check configuration in a tight loop. This setting introduces at least one check per row. I'd prefer using either a normal task writer or a delta writer depending on whether deletes are expected in the stream.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530148007
##########
File path: core/src/main/java/org/apache/iceberg/TableMetadata.java
##########
@@ -64,22 +66,25 @@ public static TableMetadata newTableMetadata(TableOperations ops,
PartitionSpec spec,
String location,
Map<String, String> properties) {
- return newTableMetadata(schema, spec, SortOrder.unsorted(), location, properties, DEFAULT_TABLE_FORMAT_VERSION);
+ int formatVersion = PropertyUtil.propertyAsInt(properties, FORMAT_VERSION, FORMAT_VERSION_DEFAULT);
+ return newTableMetadata(schema, spec, SortOrder.unsorted(), location, properties, formatVersion);
Review comment:
OK, I can change to use the `TestTables` to test the format v2. If we think all of the v2 work are finished, then we open to expose it to users.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530034825
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
+
+ default:
+ throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
+ }
+ }
+
+ @Override
+ public void close() {
+ Tasks.foreach(deltaWriterMap.values())
+ .throwFailureWhenFinished()
+ .noRetry()
+ .run(deltaWriter -> {
+ try {
+ deltaWriter.close();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ }
+
+ private class RowDataDeltaWriter extends BaseDeltaWriter {
+
+ private RowDataDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds) {
+ super(partition, equalityFieldIds, schema);
+ }
+
+ @Override
+ protected StructLike asKey(RowData row) {
+ return rowDataWrapper.wrap(row);
Review comment:
I think the tests work because the ID column is the first column in the record.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530174989
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
##########
@@ -62,7 +62,7 @@ public void open() {
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
// close all open files and emit files to downstream committer operator
- for (DataFile dataFile : writer.complete()) {
+ for (DataFile dataFile : writer.complete().dataFiles()) {
Review comment:
Yes, it should be emit the `deleteFiles` to downstream, and commit both delete files and data files into iceberg table by `RowDelta` API. I think that would be a separate PR to address this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530021908
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkFileAppenderFactory.java
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.apache.avro.io.DatumWriter;
+import org.apache.flink.table.data.RowData;
+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.DeleteUtil;
+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;
+
+public class FlinkFileAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
+ private final Schema schema;
+ private final RowType flinkSchema;
+ private final Map<String, String> props;
+ private final PartitionSpec spec;
+ private final List<Integer> equalityFieldIds;
+
+ private final boolean posDeleteWithRow;
+
+ public FlinkFileAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props,
+ PartitionSpec spec, List<Integer> equalityFieldIds) {
+ this.schema = schema;
+ this.flinkSchema = flinkSchema;
+ this.props = props;
+ this.spec = spec;
+ this.equalityFieldIds = equalityFieldIds;
+
+ // TODO Set this switch to true if needed.
+ this.posDeleteWithRow = false;
+ }
+
+ @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(flinkSchema))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(schema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkSchema, msgType))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .rowSchema(schema)
+ .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);
+ Schema rowSchema = posDeleteWithRow ? schema : null;
+ try {
+ switch (format) {
+ case AVRO:
+ Function<org.apache.avro.Schema, DatumWriter<?>> writeFunc = null;
+ if (rowSchema != null) {
+ writeFunc = ignore -> new FlinkAvroWriter(FlinkSchemaUtil.convert(rowSchema));
+ }
+
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(writeFunc)
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(rowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ RowType flinkParquetRowType = FlinkSchemaUtil.convert(DeleteUtil.posDeleteSchema(rowSchema));
Review comment:
I think we should separately fix the schema that gets passed to the `createWriterFunc`. That's a bug.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r529605950
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Comparators;
+
+public class SortedPosDeleteWriter<T> implements Closeable {
+ private static final int RECORDS_NUM_THRESHOLD = 1000_000;
+
+ private final Map<CharSequence, List<Long>> posDeletes = Maps.newHashMap();
Review comment:
Maybe we could just put the pair `<path, pos>` into a fixed array, that seems more memory efficient ?
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530184824
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestTaskWriters.java
##########
@@ -235,11 +252,115 @@ public void testRandomData() throws IOException {
}
}
+ @Test
+ public void testWriteEqualityDelete() throws IOException {
+ if (format == FileFormat.ORC) {
Review comment:
Yes, make sense !
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530034537
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestTaskWriters.java
##########
@@ -235,11 +252,115 @@ public void testRandomData() throws IOException {
}
}
+ @Test
+ public void testWriteEqualityDelete() throws IOException {
+ if (format == FileFormat.ORC) {
Review comment:
Can you use `Assume.assumeTrue`? That way it doesn't look like ORC is passing. It shows up as skipped.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530020227
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Comparators;
+
+public class SortedPosDeleteWriter<T> implements Closeable {
+ private static final int RECORDS_NUM_THRESHOLD = 1000_000;
+
+ private final Map<CharSequence, List<Long>> posDeletes = Maps.newHashMap();
Review comment:
That would be memory efficient if all of the paths that are equal are also the same reference. I like how using a map doesn't have that assumption, but it probably isn't a big concern.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530030719
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.flink.RowDataWrapper;
+import org.apache.iceberg.io.BaseTaskWriter;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.Tasks;
+
+public class RowDataTaskWriter extends BaseTaskWriter<RowData> {
+
+ private static final PartitionKey UNPARTITIONED_KEY = new PartitionKey(PartitionSpec.unpartitioned(), null);
+
+ private final Schema schema;
+ private final PartitionKey partitionKey;
+ private final RowDataWrapper rowDataWrapper;
+ private final List<Integer> equalityFieldIds;
+
+ private final Map<PartitionKey, RowDataDeltaWriter> deltaWriterMap = Maps.newHashMap();
+
+ RowDataTaskWriter(PartitionSpec spec,
+ FileFormat format,
+ FileAppenderFactory<RowData> appenderFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ RowType flinkSchema,
+ List<Integer> equalityFieldIds) {
+ super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
+
+ this.schema = schema;
+ this.equalityFieldIds = equalityFieldIds;
+
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.rowDataWrapper = new RowDataWrapper(flinkSchema, schema.asStruct());
+ }
+
+ @Override
+ public void write(RowData row) throws IOException {
+ RowDataDeltaWriter deltaWriter;
+
+ if (spec().fields().size() <= 0) {
+ // Create and cache the delta writer if absent for unpartitioned table.
+ deltaWriter = deltaWriterMap.get(UNPARTITIONED_KEY);
+ if (deltaWriter == null) {
+ deltaWriter = new RowDataDeltaWriter(null, equalityFieldIds);
+ deltaWriterMap.put(UNPARTITIONED_KEY, deltaWriter);
+ }
+ } else {
+ // Refresh the current partition key.
+ partitionKey.partition(rowDataWrapper.wrap(row));
+
+ // Create and cache the delta writer if absent for partitioned table.
+ deltaWriter = deltaWriterMap.get(partitionKey);
+ if (deltaWriter == null) {
+ // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
+ PartitionKey copiedKey = partitionKey.copy();
+ deltaWriter = new RowDataDeltaWriter(copiedKey, equalityFieldIds);
+ deltaWriterMap.put(copiedKey, deltaWriter);
+ }
+ }
+
+ switch (row.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ deltaWriter.write(row);
+ break;
+
+ case DELETE:
+ case UPDATE_BEFORE:
+ deltaWriter.delete(row);
+ break;
+
+ default:
+ throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
+ }
+ }
+
+ @Override
+ public void close() {
+ Tasks.foreach(deltaWriterMap.values())
+ .throwFailureWhenFinished()
+ .noRetry()
+ .run(deltaWriter -> {
+ try {
+ deltaWriter.close();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ }
+
+ private class RowDataDeltaWriter extends BaseDeltaWriter {
+
+ private RowDataDeltaWriter(PartitionKey partition, List<Integer> equalityFieldIds) {
+ super(partition, equalityFieldIds, schema);
+ }
+
+ @Override
+ protected StructLike asKey(RowData row) {
+ return rowDataWrapper.wrap(row);
Review comment:
It doesn't look like this projects values. Shouldn't this call `partitionKey.partition(rowDataWrapper.wrap(row))`?
The fanout writer used this:
```
@Override
protected PartitionKey partition(RowData row) {
partitionKey.partition(rowDataWrapper.wrap(row));
return partitionKey;
}
```
----------------------------------------------------------------
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 pull request #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#issuecomment-733380672
@openinx, this looks great to me!
I found a few issues to address, but I think the general design and structure is good. Should we start breaking it into smaller commits to get the changes into master?
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530029220
##########
File path: spark/src/main/java/org/apache/iceberg/spark/source/SparkAppenderFactory.java
##########
@@ -86,4 +94,24 @@
throw new RuntimeIOException(e);
}
}
+
+ @Override
+ public DataWriter<InternalRow> newDataWriter(EncryptedOutputFile file, FileFormat fileFormat,
+ StructLike partition) {
+ return new DataWriter<>(
+ newAppender(file.encryptingOutputFile(), fileFormat), fileFormat,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
+
+ @Override
+ public EqualityDeleteWriter<InternalRow> newEqDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
Review comment:
I think one of the first commits to get this in could make these small changes to both Spark and Flink to update the appender factory. We could do the same for generics 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] openinx commented on pull request #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#issuecomment-750833431
All this work has been merged in several individual PR, I plan to close this PR now.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530020823
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Comparators;
+
+public class SortedPosDeleteWriter<T> implements Closeable {
+ private static final int RECORDS_NUM_THRESHOLD = 1000_000;
+
+ private final Map<CharSequence, List<Long>> posDeletes = Maps.newHashMap();
Review comment:
This needs to use a new `CharSequenceMap` or `Map<CharSequenceWrapper, List<Long>>`.
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530019646
##########
File path: core/src/main/java/org/apache/iceberg/util/StructLikeMap.java
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.util;
+
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Types;
+
+public class StructLikeMap<T> extends AbstractMap<StructLike, T> implements Map<StructLike, T> {
+
+ public static <T> StructLikeMap<T> create(Types.StructType type) {
+ return new StructLikeMap<>(type);
+ }
+
+ private final Types.StructType type;
+ private final Map<StructLikeWrapper, T> wrapperMap;
+ private final ThreadLocal<StructLikeWrapper> wrappers;
+
+ private StructLikeMap(Types.StructType type) {
+ this.type = type;
+ this.wrapperMap = Maps.newHashMap();
+ this.wrappers = ThreadLocal.withInitial(() -> StructLikeWrapper.forType(type));
+ }
+
+ @Override
+ public int size() {
+ return wrapperMap.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return wrapperMap.isEmpty();
+ }
+
+ @Override
+ public boolean containsKey(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ boolean result = wrapperMap.containsKey(wrapper.set((StructLike) key));
+ wrapper.set(null);
+ return result;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean containsValue(Object value) {
+ throw new UnsupportedOperationException();
Review comment:
Can't this just delegate?
```java
@Override
public boolean containsValue(Object value) {
return wrapperMap.containsValue(value);
}
```
----------------------------------------------------------------
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 #1818: Flink: Add RowDataTaskWriter to accept equality deletions.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1818:
URL: https://github.com/apache/iceberg/pull/1818#discussion_r530018658
##########
File path: core/src/main/java/org/apache/iceberg/util/StructLikeMap.java
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.util;
+
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Types;
+
+public class StructLikeMap<T> extends AbstractMap<StructLike, T> implements Map<StructLike, T> {
+
+ public static <T> StructLikeMap<T> create(Types.StructType type) {
+ return new StructLikeMap<>(type);
+ }
+
+ private final Types.StructType type;
+ private final Map<StructLikeWrapper, T> wrapperMap;
+ private final ThreadLocal<StructLikeWrapper> wrappers;
+
+ private StructLikeMap(Types.StructType type) {
+ this.type = type;
+ this.wrapperMap = Maps.newHashMap();
+ this.wrappers = ThreadLocal.withInitial(() -> StructLikeWrapper.forType(type));
+ }
+
+ @Override
+ public int size() {
+ return wrapperMap.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return wrapperMap.isEmpty();
+ }
+
+ @Override
+ public boolean containsKey(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ boolean result = wrapperMap.containsKey(wrapper.set((StructLike) key));
+ wrapper.set(null);
+ return result;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean containsValue(Object value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public T get(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ T value = wrapperMap.get(wrapper.set((StructLike) key));
+ wrapper.set(null);
+ return value;
+ }
+ return null;
+ }
+
+ @Override
+ public T put(StructLike key, T value) {
+ return wrapperMap.put(StructLikeWrapper.forType(type).set(key), value);
+ }
+
+ @Override
+ public T remove(Object key) {
+ if (key instanceof StructLike) {
+ StructLikeWrapper wrapper = wrappers.get();
+ T value = wrapperMap.remove(wrapper.set((StructLike) key));
+ wrapper.set(null); // don't hold a reference to the value.
+ return value;
+ }
+ return null;
+ }
+
+ @Override
+ public void putAll(Map<? extends StructLike, ? extends T> keyValues) {
+ if (keyValues != null && !keyValues.isEmpty()) {
+ for (Map.Entry<? extends StructLike, ? extends T> pair : keyValues.entrySet()) {
+ wrapperMap.put(StructLikeWrapper.forType(type).set(pair.getKey()), pair.getValue());
+ }
+ }
+ }
+
+ @Override
+ public void clear() {
+ wrapperMap.clear();
+ }
+
+ @Override
+ public Set<StructLike> keySet() {
+ return wrapperMap.keySet().stream().map(StructLikeWrapper::get).collect(Collectors.toSet());
Review comment:
I think this should use a `StructLikeSet`, or else there is no expectation that the returned set will function properly.
----------------------------------------------------------------
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