diff --git a/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java b/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java index e1bea38a3843..9c197be5022e 100644 --- a/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java +++ b/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java @@ -690,7 +690,7 @@ default PositionDeleteWriter writer( } /** - * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. Use {@link * #writer(OutputFile, FileFormat, PartitionSpec, StructLike)} instead. */ diff --git a/core/src/main/java/org/apache/iceberg/deletes/PositionDelete.java b/core/src/main/java/org/apache/iceberg/deletes/PositionDelete.java index c3b6cbaa9bff..8893d2a243a0 100644 --- a/core/src/main/java/org/apache/iceberg/deletes/PositionDelete.java +++ b/core/src/main/java/org/apache/iceberg/deletes/PositionDelete.java @@ -44,7 +44,7 @@ public PositionDelete set(CharSequence newPath, long newPos) { } /** - * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. Use {@link * #set(CharSequence, long)} instead. */ @@ -70,7 +70,7 @@ public long pos() { } /** - * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ @Deprecated diff --git a/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java b/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java index ea3e3896c0ec..7c53ea09a1d6 100644 --- a/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java +++ b/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java @@ -63,7 +63,7 @@ public class PositionDeleteWriter implements FileWriter, De /** * Creates a new position delete writer. * - * @deprecated since 1.11.0, will be updated in 1.12.0 to accept {@code + * @deprecated since 1.11.0, will be updated in 1.13.0 to accept {@code * FileAppender>} instead of {@code FileAppender}. */ @Deprecated diff --git a/core/src/main/java/org/apache/iceberg/io/PositionDeltaWriter.java b/core/src/main/java/org/apache/iceberg/io/PositionDeltaWriter.java index 8ce4e64c3a8b..88b40ca60532 100644 --- a/core/src/main/java/org/apache/iceberg/io/PositionDeltaWriter.java +++ b/core/src/main/java/org/apache/iceberg/io/PositionDeltaWriter.java @@ -77,7 +77,11 @@ default void delete(CharSequence path, long pos, PartitionSpec spec, StructLike * @param row a deleted row * @param spec a partition spec * @param partition a partition or null if the spec is unpartitioned + * @deprecated since 1.12.0, will be removed in 1.13.0; position deletes that include row data are + * no longer supported. Use {@link #delete(CharSequence, long, PartitionSpec, StructLike)} + * instead, which will become the method implementations override. */ + @Deprecated void delete(CharSequence path, long pos, T row, PartitionSpec spec, StructLike partition); /** diff --git a/data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java b/data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java index 914dc3672b77..d07d6f24cc32 100644 --- a/data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java +++ b/data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java @@ -79,7 +79,7 @@ public class GenericFileWriterFactory extends RegistryBasedFileWriterFactory extends TestBase { - private static final int FORMAT_V2 = 2; - - private PartitionKey partition = null; - private OutputFileFactory fileFactory = null; - - @Parameter(index = 1) - protected FileFormat format; - - @Parameter(index = 2) - private boolean partitioned; - - @Parameters(name = "formatVersion = {0}, FileFormat={1}, partitioned={2}") - protected static List parameters() { - return Arrays.asList( - new Object[] {FORMAT_V2, FileFormat.AVRO, false}, - new Object[] {FORMAT_V2, FileFormat.AVRO, true}, - new Object[] {FORMAT_V2, FileFormat.ORC, false}, - new Object[] {FORMAT_V2, FileFormat.ORC, true}, - new Object[] {FORMAT_V2, FileFormat.PARQUET, false}, - new Object[] {FORMAT_V2, FileFormat.PARQUET, true}); - } - - @Override - @BeforeEach - public void setupTable() throws Exception { - this.metadataDir = new File(tableDir, "metadata"); - - if (partitioned) { - this.table = create(SCHEMA, SPEC); - } else { - this.table = create(SCHEMA, PartitionSpec.unpartitioned()); - } - this.partition = createPartitionKey(); - this.fileFactory = OutputFileFactory.builderFor(table, 1, 1).format(format).build(); - - table.updateProperties().defaultFormat(format).commit(); - } - - protected abstract FileAppenderFactory createAppenderFactory( - List equalityFieldIds, Schema eqDeleteSchema, Schema posDeleteRowSchema); - - protected abstract T createRow(Integer id, String data); - - protected abstract StructLikeSet expectedRowSet(Iterable records) throws IOException; - - private StructLikeSet actualRowSet(String... columns) throws IOException { - StructLikeSet set = StructLikeSet.create(table.schema().asStruct()); - try (CloseableIterable reader = IcebergGenerics.read(table).select(columns).build()) { - reader.forEach(set::add); - } - return set; - } - - private PartitionKey createPartitionKey() { - if (table.spec().isUnpartitioned()) { - return null; - } - - Record record = GenericRecord.create(table.schema()).copy(ImmutableMap.of("data", "aaa")); - - PartitionKey partitionKey = new PartitionKey(table.spec(), table.schema()); - partitionKey.partition(record); - - return partitionKey; - } - - private EncryptedOutputFile createEncryptedOutputFile() { - if (partition == null) { - return fileFactory.newOutputFile(); - } else { - return fileFactory.newOutputFile(partition); - } - } - - private List testRowSet() { - return Lists.newArrayList( - createRow(1, "aaa"), - createRow(2, "bbb"), - createRow(3, "ccc"), - createRow(4, "ddd"), - createRow(5, "eee")); - } - - private DataFile prepareDataFile(List rowSet, FileAppenderFactory appenderFactory) - throws IOException { - DataWriter writer = - appenderFactory.newDataWriter(createEncryptedOutputFile(), format, partition); - try (DataWriter closeableWriter = writer) { - for (T row : rowSet) { - closeableWriter.write(row); - } - } - - return writer.toDataFile(); - } - - @TestTemplate - public void testDataWriter() throws IOException { - FileAppenderFactory appenderFactory = createAppenderFactory(null, null, null); - - List rowSet = testRowSet(); - DataFile dataFile = prepareDataFile(rowSet, appenderFactory); - - table.newRowDelta().addRows(dataFile).commit(); - - assertThat(actualRowSet("*")) - .as("Should have the expected records.") - .isEqualTo(expectedRowSet(rowSet)); - } - - @TestTemplate - public void testEqDeleteWriter() throws IOException { - List equalityFieldIds = Lists.newArrayList(table.schema().findField("id").fieldId()); - Schema eqDeleteRowSchema = table.schema().select("id"); - FileAppenderFactory appenderFactory = - createAppenderFactory(equalityFieldIds, eqDeleteRowSchema, null); - - List rowSet = testRowSet(); - DataFile dataFile = prepareDataFile(rowSet, appenderFactory); - - table.newRowDelta().addRows(dataFile).commit(); - - // The equality field is 'id'. No matter what the value of 'data' field is, we should delete - // the 1th, 3th, 5th - // rows. - List deletes = - Lists.newArrayList(createRow(1, "aaa"), createRow(3, "bbb"), createRow(5, "ccc")); - EncryptedOutputFile out = createEncryptedOutputFile(); - EqualityDeleteWriter eqDeleteWriter = - appenderFactory.newEqDeleteWriter(out, format, partition); - try (EqualityDeleteWriter closeableWriter = eqDeleteWriter) { - closeableWriter.write(deletes); - } - - // Check that the delete equality file has the expected equality deletes. - GenericRecord gRecord = GenericRecord.create(eqDeleteRowSchema); - Set expectedDeletes = - Sets.newHashSet(gRecord.copy("id", 1), gRecord.copy("id", 3), gRecord.copy("id", 5)); - assertThat( - Sets.newHashSet( - createReader(eqDeleteRowSchema, out.encryptingOutputFile().toInputFile()))) - .isEqualTo(expectedDeletes); - - table.newRowDelta().addDeletes(eqDeleteWriter.toDeleteFile()).commit(); - - List expected = Lists.newArrayList(createRow(2, "bbb"), createRow(4, "ddd")); - assertThat(actualRowSet("*")) - .as("Should have the expected records") - .isEqualTo(expectedRowSet(expected)); - } - - @TestTemplate - public void testPosDeleteWriter() throws IOException { - // Initialize FileAppenderFactory without pos-delete row schema. - FileAppenderFactory appenderFactory = createAppenderFactory(null, null, null); - - List rowSet = testRowSet(); - DataFile dataFile = prepareDataFile(rowSet, appenderFactory); - - List> deletes = - Lists.newArrayList( - Pair.of(dataFile.location(), 0L), - Pair.of(dataFile.location(), 2L), - Pair.of(dataFile.location(), 4L)); - - EncryptedOutputFile out = createEncryptedOutputFile(); - PositionDeleteWriter eqDeleteWriter = - appenderFactory.newPosDeleteWriter(out, format, partition); - PositionDelete posDelete = PositionDelete.create(); - try (PositionDeleteWriter closeableWriter = eqDeleteWriter) { - for (Pair delete : deletes) { - closeableWriter.write(posDelete.set(delete.first(), delete.second(), null)); - } - } - - // Check that the pos delete file has the expected pos deletes. - Schema pathPosSchema = DeleteSchemaUtil.pathPosSchema(); - GenericRecord gRecord = GenericRecord.create(pathPosSchema); - Set expectedDeletes = - Sets.newHashSet( - gRecord.copy("file_path", dataFile.location(), "pos", 0L), - gRecord.copy("file_path", dataFile.location(), "pos", 2L), - gRecord.copy("file_path", dataFile.location(), "pos", 4L)); - assertThat( - Sets.newHashSet(createReader(pathPosSchema, out.encryptingOutputFile().toInputFile()))) - .isEqualTo(expectedDeletes); - - table - .newRowDelta() - .addRows(dataFile) - .addDeletes(eqDeleteWriter.toDeleteFile()) - .validateDataFilesExist(eqDeleteWriter.referencedDataFiles()) - .validateDeletedFiles() - .commit(); - - List expected = Lists.newArrayList(createRow(2, "bbb"), createRow(4, "ddd")); - assertThat(actualRowSet("*")) - .as("Should have the expected records") - .isEqualTo(expectedRowSet(expected)); - } - - @TestTemplate - public void testPosDeleteWriterWithRowSchema() throws IOException { - FileAppenderFactory appenderFactory = createAppenderFactory(null, null, table.schema()); - - List rowSet = testRowSet(); - DataFile dataFile = prepareDataFile(rowSet, appenderFactory); - - List> deletes = - Lists.newArrayList( - positionDelete(dataFile.location(), 0, rowSet.get(0)), - positionDelete(dataFile.location(), 2, rowSet.get(2)), - positionDelete(dataFile.location(), 4, rowSet.get(4))); - - EncryptedOutputFile out = createEncryptedOutputFile(); - PositionDeleteWriter eqDeleteWriter = - appenderFactory.newPosDeleteWriter(out, format, partition); - PositionDelete posDelete = PositionDelete.create(); - try (PositionDeleteWriter closeableWriter = eqDeleteWriter) { - for (PositionDelete delete : deletes) { - closeableWriter.write(posDelete.set(delete.path(), delete.pos(), delete.row())); - } - } - - // Check that the pos delete file has the expected pos deletes. - Schema pathPosRowSchema = DeleteSchemaUtil.posDeleteSchema(table.schema()); - GenericRecord gRecord = GenericRecord.create(pathPosRowSchema); - GenericRecord rowRecord = GenericRecord.create(table.schema()); - Set expectedDeletes = - Sets.newHashSet( - gRecord.copy( - "file_path", - dataFile.location(), - "pos", - 0L, - "row", - rowRecord.copy("id", 1, "data", "aaa")), - gRecord.copy( - "file_path", - dataFile.location(), - "pos", - 2L, - "row", - rowRecord.copy("id", 3, "data", "ccc")), - gRecord.copy( - "file_path", - dataFile.location(), - "pos", - 4L, - "row", - rowRecord.copy("id", 5, "data", "eee"))); - assertThat( - Sets.newHashSet( - createReader(pathPosRowSchema, out.encryptingOutputFile().toInputFile()))) - .isEqualTo(expectedDeletes); - - table - .newRowDelta() - .addRows(dataFile) - .addDeletes(eqDeleteWriter.toDeleteFile()) - .validateDataFilesExist(eqDeleteWriter.referencedDataFiles()) - .validateDeletedFiles() - .commit(); - - List expected = Lists.newArrayList(createRow(2, "bbb"), createRow(4, "ddd")); - assertThat(actualRowSet("*")) - .as("Should have the expected records") - .isEqualTo(expectedRowSet(expected)); - } - - private CloseableIterable createReader(Schema schema, InputFile inputFile) { - switch (format) { - case PARQUET: - return Parquet.read(inputFile) - .project(schema) - .createReaderFunc(fileSchema -> GenericParquetReaders.buildReader(schema, fileSchema)) - .build(); - - case AVRO: - return Avro.read(inputFile) - .project(schema) - .createResolvingReader(PlannedDataReader::create) - .build(); - - case ORC: - return ORC.read(inputFile) - .project(schema) - .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(schema, fileSchema)) - .build(); - - default: - throw new UnsupportedOperationException("Unsupported file format: " + format); - } - } -} diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java deleted file mode 100644 index 85dcc3133bd6..000000000000 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java +++ /dev/null @@ -1,296 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.iceberg.flink.sink; - -import java.io.IOException; -import java.io.Serializable; -import java.io.UncheckedIOException; -import java.util.Map; -import org.apache.flink.table.data.RowData; -import org.apache.flink.table.data.StringData; -import org.apache.flink.table.types.logical.RowType; -import org.apache.iceberg.FileFormat; -import org.apache.iceberg.MetricsConfig; -import org.apache.iceberg.PartitionSpec; -import org.apache.iceberg.Schema; -import org.apache.iceberg.StructLike; -import org.apache.iceberg.Table; -import org.apache.iceberg.avro.Avro; -import org.apache.iceberg.deletes.EqualityDeleteWriter; -import org.apache.iceberg.deletes.PositionDeleteWriter; -import org.apache.iceberg.encryption.EncryptedOutputFile; -import org.apache.iceberg.flink.FlinkSchemaUtil; -import org.apache.iceberg.flink.data.FlinkAvroWriter; -import org.apache.iceberg.flink.data.FlinkOrcWriter; -import org.apache.iceberg.flink.data.FlinkParquetWriters; -import org.apache.iceberg.io.DataWriter; -import org.apache.iceberg.io.DeleteSchemaUtil; -import org.apache.iceberg.io.FileAppender; -import org.apache.iceberg.io.FileAppenderFactory; -import org.apache.iceberg.io.OutputFile; -import org.apache.iceberg.orc.ORC; -import org.apache.iceberg.parquet.Parquet; -import org.apache.iceberg.relocated.com.google.common.base.Preconditions; - -/** - * @deprecated Deprecated as of 1.11.0 in favor of {@link FlinkFileWriterFactory}. This class will - * be removed in the 1.12.0. - */ -@Deprecated -public class FlinkAppenderFactory implements FileAppenderFactory, Serializable { - private final Schema schema; - private final RowType flinkSchema; - private final Map props; - private final PartitionSpec spec; - private final int[] equalityFieldIds; - private final Schema eqDeleteRowSchema; - private final Schema posDeleteRowSchema; - private final Table table; - - private RowType eqDeleteFlinkSchema = null; - private RowType posDeleteFlinkSchema = null; - - public FlinkAppenderFactory( - Table table, - Schema schema, - RowType flinkSchema, - Map props, - PartitionSpec spec, - int[] equalityFieldIds, - Schema eqDeleteRowSchema) { - this(table, schema, flinkSchema, props, spec, equalityFieldIds, eqDeleteRowSchema, null); - } - - /** - * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.12.0. - * Position deletes that include row data are no longer supported. Use {@link - * #FlinkAppenderFactory(Table, Schema, RowType, Map, PartitionSpec, int[], Schema)} instead. - */ - @Deprecated - public FlinkAppenderFactory( - Table table, - Schema schema, - RowType flinkSchema, - Map props, - PartitionSpec spec, - int[] equalityFieldIds, - Schema eqDeleteRowSchema, - Schema posDeleteRowSchema) { - Preconditions.checkNotNull(table, "Table shouldn't be null"); - this.table = table; - this.schema = schema; - this.flinkSchema = flinkSchema; - this.props = props; - this.spec = spec; - this.equalityFieldIds = equalityFieldIds; - this.eqDeleteRowSchema = eqDeleteRowSchema; - this.posDeleteRowSchema = posDeleteRowSchema; - } - - private RowType lazyEqDeleteFlinkSchema() { - if (eqDeleteFlinkSchema == null) { - Preconditions.checkNotNull(eqDeleteRowSchema, "Equality delete row schema shouldn't be null"); - this.eqDeleteFlinkSchema = FlinkSchemaUtil.convert(eqDeleteRowSchema); - } - return eqDeleteFlinkSchema; - } - - private RowType lazyPosDeleteFlinkSchema() { - if (posDeleteFlinkSchema == null) { - Preconditions.checkNotNull(posDeleteRowSchema, "Pos-delete row schema shouldn't be null"); - this.posDeleteFlinkSchema = FlinkSchemaUtil.convert(posDeleteRowSchema); - } - return this.posDeleteFlinkSchema; - } - - @Override - public FileAppender newAppender(OutputFile outputFile, FileFormat format) { - MetricsConfig metricsConfig = MetricsConfig.forTable(table); - try { - switch (format) { - case AVRO: - return Avro.write(outputFile) - .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema)) - .setAll(props) - .schema(schema) - .metricsConfig(metricsConfig) - .overwrite() - .build(); - - case ORC: - return ORC.write(outputFile) - .createWriterFunc( - (iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema)) - .setAll(props) - .metricsConfig(metricsConfig) - .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 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 newEqDeleteWriter( - EncryptedOutputFile outputFile, FileFormat format, StructLike partition) { - Preconditions.checkState( - equalityFieldIds != null && equalityFieldIds.length > 0, - "Equality field ids shouldn't be null or empty when creating equality-delete writer"); - Preconditions.checkNotNull( - eqDeleteRowSchema, - "Equality delete row schema shouldn't be null when creating equality-delete writer"); - - MetricsConfig metricsConfig = MetricsConfig.forTable(table); - try { - switch (format) { - case AVRO: - return Avro.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc(ignore -> new FlinkAvroWriter(lazyEqDeleteFlinkSchema())) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(eqDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .equalityFieldIds(equalityFieldIds) - .buildEqualityWriter(); - - case ORC: - return ORC.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - (iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(eqDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .equalityFieldIds(equalityFieldIds) - .buildEqualityWriter(); - - case PARQUET: - return Parquet.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - msgType -> FlinkParquetWriters.buildWriter(lazyEqDeleteFlinkSchema(), msgType)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(eqDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .equalityFieldIds(equalityFieldIds) - .buildEqualityWriter(); - - default: - throw new UnsupportedOperationException( - "Cannot write equality-deletes for unsupported file format: " + format); - } - } catch (IOException e) { - throw new UncheckedIOException(e); - } - } - - @Override - public PositionDeleteWriter newPosDeleteWriter( - EncryptedOutputFile outputFile, FileFormat format, StructLike partition) { - MetricsConfig metricsConfig = MetricsConfig.forPositionDelete(); - try { - switch (format) { - case AVRO: - return Avro.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc(ignore -> new FlinkAvroWriter(lazyPosDeleteFlinkSchema())) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(posDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .buildPositionWriter(); - - case ORC: - RowType orcPosDeleteSchema = - FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema)); - return ORC.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - (iSchema, typDesc) -> FlinkOrcWriter.buildWriter(orcPosDeleteSchema, iSchema)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(posDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .transformPaths(path -> StringData.fromString(path.toString())) - .buildPositionWriter(); - - case PARQUET: - RowType flinkPosDeleteSchema = - FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema)); - return Parquet.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - msgType -> FlinkParquetWriters.buildWriter(flinkPosDeleteSchema, msgType)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(posDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .transformPaths(path -> StringData.fromString(path.toString())) - .buildPositionWriter(); - - default: - throw new UnsupportedOperationException( - "Cannot write pos-deletes for unsupported file format: " + format); - } - } catch (IOException e) { - throw new UncheckedIOException(e); - } - } -} diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java deleted file mode 100644 index dd89f43483b0..000000000000 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * 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.util.List; -import org.apache.flink.table.data.RowData; -import org.apache.flink.table.types.logical.RowType; -import org.apache.iceberg.Schema; -import org.apache.iceberg.flink.FlinkSchemaUtil; -import org.apache.iceberg.flink.RowDataWrapper; -import org.apache.iceberg.flink.SimpleDataUtil; -import org.apache.iceberg.io.FileAppenderFactory; -import org.apache.iceberg.io.TestAppenderFactory; -import org.apache.iceberg.util.ArrayUtil; -import org.apache.iceberg.util.StructLikeSet; - -public class TestFlinkAppenderFactory extends TestAppenderFactory { - - private final RowType rowType = FlinkSchemaUtil.convert(SCHEMA); - - @Override - protected FileAppenderFactory createAppenderFactory( - List equalityFieldIds, Schema eqDeleteSchema, Schema posDeleteRowSchema) { - return new FlinkAppenderFactory( - table, - table.schema(), - rowType, - table.properties(), - table.spec(), - ArrayUtil.toIntArray(equalityFieldIds), - eqDeleteSchema, - posDeleteRowSchema); - } - - @Override - protected RowData createRow(Integer id, String data) { - return SimpleDataUtil.createRowData(id, data); - } - - @Override - protected StructLikeSet expectedRowSet(Iterable rows) { - StructLikeSet set = StructLikeSet.create(table.schema().asStruct()); - for (RowData row : rows) { - RowDataWrapper wrapper = new RowDataWrapper(rowType, table.schema().asStruct()); - set.add(wrapper.wrap(row)); - } - return set; - } -} diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java deleted file mode 100644 index 85dcc3133bd6..000000000000 --- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java +++ /dev/null @@ -1,296 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.iceberg.flink.sink; - -import java.io.IOException; -import java.io.Serializable; -import java.io.UncheckedIOException; -import java.util.Map; -import org.apache.flink.table.data.RowData; -import org.apache.flink.table.data.StringData; -import org.apache.flink.table.types.logical.RowType; -import org.apache.iceberg.FileFormat; -import org.apache.iceberg.MetricsConfig; -import org.apache.iceberg.PartitionSpec; -import org.apache.iceberg.Schema; -import org.apache.iceberg.StructLike; -import org.apache.iceberg.Table; -import org.apache.iceberg.avro.Avro; -import org.apache.iceberg.deletes.EqualityDeleteWriter; -import org.apache.iceberg.deletes.PositionDeleteWriter; -import org.apache.iceberg.encryption.EncryptedOutputFile; -import org.apache.iceberg.flink.FlinkSchemaUtil; -import org.apache.iceberg.flink.data.FlinkAvroWriter; -import org.apache.iceberg.flink.data.FlinkOrcWriter; -import org.apache.iceberg.flink.data.FlinkParquetWriters; -import org.apache.iceberg.io.DataWriter; -import org.apache.iceberg.io.DeleteSchemaUtil; -import org.apache.iceberg.io.FileAppender; -import org.apache.iceberg.io.FileAppenderFactory; -import org.apache.iceberg.io.OutputFile; -import org.apache.iceberg.orc.ORC; -import org.apache.iceberg.parquet.Parquet; -import org.apache.iceberg.relocated.com.google.common.base.Preconditions; - -/** - * @deprecated Deprecated as of 1.11.0 in favor of {@link FlinkFileWriterFactory}. This class will - * be removed in the 1.12.0. - */ -@Deprecated -public class FlinkAppenderFactory implements FileAppenderFactory, Serializable { - private final Schema schema; - private final RowType flinkSchema; - private final Map props; - private final PartitionSpec spec; - private final int[] equalityFieldIds; - private final Schema eqDeleteRowSchema; - private final Schema posDeleteRowSchema; - private final Table table; - - private RowType eqDeleteFlinkSchema = null; - private RowType posDeleteFlinkSchema = null; - - public FlinkAppenderFactory( - Table table, - Schema schema, - RowType flinkSchema, - Map props, - PartitionSpec spec, - int[] equalityFieldIds, - Schema eqDeleteRowSchema) { - this(table, schema, flinkSchema, props, spec, equalityFieldIds, eqDeleteRowSchema, null); - } - - /** - * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.12.0. - * Position deletes that include row data are no longer supported. Use {@link - * #FlinkAppenderFactory(Table, Schema, RowType, Map, PartitionSpec, int[], Schema)} instead. - */ - @Deprecated - public FlinkAppenderFactory( - Table table, - Schema schema, - RowType flinkSchema, - Map props, - PartitionSpec spec, - int[] equalityFieldIds, - Schema eqDeleteRowSchema, - Schema posDeleteRowSchema) { - Preconditions.checkNotNull(table, "Table shouldn't be null"); - this.table = table; - this.schema = schema; - this.flinkSchema = flinkSchema; - this.props = props; - this.spec = spec; - this.equalityFieldIds = equalityFieldIds; - this.eqDeleteRowSchema = eqDeleteRowSchema; - this.posDeleteRowSchema = posDeleteRowSchema; - } - - private RowType lazyEqDeleteFlinkSchema() { - if (eqDeleteFlinkSchema == null) { - Preconditions.checkNotNull(eqDeleteRowSchema, "Equality delete row schema shouldn't be null"); - this.eqDeleteFlinkSchema = FlinkSchemaUtil.convert(eqDeleteRowSchema); - } - return eqDeleteFlinkSchema; - } - - private RowType lazyPosDeleteFlinkSchema() { - if (posDeleteFlinkSchema == null) { - Preconditions.checkNotNull(posDeleteRowSchema, "Pos-delete row schema shouldn't be null"); - this.posDeleteFlinkSchema = FlinkSchemaUtil.convert(posDeleteRowSchema); - } - return this.posDeleteFlinkSchema; - } - - @Override - public FileAppender newAppender(OutputFile outputFile, FileFormat format) { - MetricsConfig metricsConfig = MetricsConfig.forTable(table); - try { - switch (format) { - case AVRO: - return Avro.write(outputFile) - .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema)) - .setAll(props) - .schema(schema) - .metricsConfig(metricsConfig) - .overwrite() - .build(); - - case ORC: - return ORC.write(outputFile) - .createWriterFunc( - (iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema)) - .setAll(props) - .metricsConfig(metricsConfig) - .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 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 newEqDeleteWriter( - EncryptedOutputFile outputFile, FileFormat format, StructLike partition) { - Preconditions.checkState( - equalityFieldIds != null && equalityFieldIds.length > 0, - "Equality field ids shouldn't be null or empty when creating equality-delete writer"); - Preconditions.checkNotNull( - eqDeleteRowSchema, - "Equality delete row schema shouldn't be null when creating equality-delete writer"); - - MetricsConfig metricsConfig = MetricsConfig.forTable(table); - try { - switch (format) { - case AVRO: - return Avro.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc(ignore -> new FlinkAvroWriter(lazyEqDeleteFlinkSchema())) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(eqDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .equalityFieldIds(equalityFieldIds) - .buildEqualityWriter(); - - case ORC: - return ORC.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - (iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(eqDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .equalityFieldIds(equalityFieldIds) - .buildEqualityWriter(); - - case PARQUET: - return Parquet.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - msgType -> FlinkParquetWriters.buildWriter(lazyEqDeleteFlinkSchema(), msgType)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(eqDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .equalityFieldIds(equalityFieldIds) - .buildEqualityWriter(); - - default: - throw new UnsupportedOperationException( - "Cannot write equality-deletes for unsupported file format: " + format); - } - } catch (IOException e) { - throw new UncheckedIOException(e); - } - } - - @Override - public PositionDeleteWriter newPosDeleteWriter( - EncryptedOutputFile outputFile, FileFormat format, StructLike partition) { - MetricsConfig metricsConfig = MetricsConfig.forPositionDelete(); - try { - switch (format) { - case AVRO: - return Avro.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc(ignore -> new FlinkAvroWriter(lazyPosDeleteFlinkSchema())) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(posDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .buildPositionWriter(); - - case ORC: - RowType orcPosDeleteSchema = - FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema)); - return ORC.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - (iSchema, typDesc) -> FlinkOrcWriter.buildWriter(orcPosDeleteSchema, iSchema)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(posDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .transformPaths(path -> StringData.fromString(path.toString())) - .buildPositionWriter(); - - case PARQUET: - RowType flinkPosDeleteSchema = - FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema)); - return Parquet.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - msgType -> FlinkParquetWriters.buildWriter(flinkPosDeleteSchema, msgType)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(posDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .transformPaths(path -> StringData.fromString(path.toString())) - .buildPositionWriter(); - - default: - throw new UnsupportedOperationException( - "Cannot write pos-deletes for unsupported file format: " + format); - } - } catch (IOException e) { - throw new UncheckedIOException(e); - } - } -} diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java deleted file mode 100644 index dd89f43483b0..000000000000 --- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * 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.util.List; -import org.apache.flink.table.data.RowData; -import org.apache.flink.table.types.logical.RowType; -import org.apache.iceberg.Schema; -import org.apache.iceberg.flink.FlinkSchemaUtil; -import org.apache.iceberg.flink.RowDataWrapper; -import org.apache.iceberg.flink.SimpleDataUtil; -import org.apache.iceberg.io.FileAppenderFactory; -import org.apache.iceberg.io.TestAppenderFactory; -import org.apache.iceberg.util.ArrayUtil; -import org.apache.iceberg.util.StructLikeSet; - -public class TestFlinkAppenderFactory extends TestAppenderFactory { - - private final RowType rowType = FlinkSchemaUtil.convert(SCHEMA); - - @Override - protected FileAppenderFactory createAppenderFactory( - List equalityFieldIds, Schema eqDeleteSchema, Schema posDeleteRowSchema) { - return new FlinkAppenderFactory( - table, - table.schema(), - rowType, - table.properties(), - table.spec(), - ArrayUtil.toIntArray(equalityFieldIds), - eqDeleteSchema, - posDeleteRowSchema); - } - - @Override - protected RowData createRow(Integer id, String data) { - return SimpleDataUtil.createRowData(id, data); - } - - @Override - protected StructLikeSet expectedRowSet(Iterable rows) { - StructLikeSet set = StructLikeSet.create(table.schema().asStruct()); - for (RowData row : rows) { - RowDataWrapper wrapper = new RowDataWrapper(rowType, table.schema().asStruct()); - set.add(wrapper.wrap(row)); - } - return set; - } -} diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java deleted file mode 100644 index 85dcc3133bd6..000000000000 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java +++ /dev/null @@ -1,296 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.iceberg.flink.sink; - -import java.io.IOException; -import java.io.Serializable; -import java.io.UncheckedIOException; -import java.util.Map; -import org.apache.flink.table.data.RowData; -import org.apache.flink.table.data.StringData; -import org.apache.flink.table.types.logical.RowType; -import org.apache.iceberg.FileFormat; -import org.apache.iceberg.MetricsConfig; -import org.apache.iceberg.PartitionSpec; -import org.apache.iceberg.Schema; -import org.apache.iceberg.StructLike; -import org.apache.iceberg.Table; -import org.apache.iceberg.avro.Avro; -import org.apache.iceberg.deletes.EqualityDeleteWriter; -import org.apache.iceberg.deletes.PositionDeleteWriter; -import org.apache.iceberg.encryption.EncryptedOutputFile; -import org.apache.iceberg.flink.FlinkSchemaUtil; -import org.apache.iceberg.flink.data.FlinkAvroWriter; -import org.apache.iceberg.flink.data.FlinkOrcWriter; -import org.apache.iceberg.flink.data.FlinkParquetWriters; -import org.apache.iceberg.io.DataWriter; -import org.apache.iceberg.io.DeleteSchemaUtil; -import org.apache.iceberg.io.FileAppender; -import org.apache.iceberg.io.FileAppenderFactory; -import org.apache.iceberg.io.OutputFile; -import org.apache.iceberg.orc.ORC; -import org.apache.iceberg.parquet.Parquet; -import org.apache.iceberg.relocated.com.google.common.base.Preconditions; - -/** - * @deprecated Deprecated as of 1.11.0 in favor of {@link FlinkFileWriterFactory}. This class will - * be removed in the 1.12.0. - */ -@Deprecated -public class FlinkAppenderFactory implements FileAppenderFactory, Serializable { - private final Schema schema; - private final RowType flinkSchema; - private final Map props; - private final PartitionSpec spec; - private final int[] equalityFieldIds; - private final Schema eqDeleteRowSchema; - private final Schema posDeleteRowSchema; - private final Table table; - - private RowType eqDeleteFlinkSchema = null; - private RowType posDeleteFlinkSchema = null; - - public FlinkAppenderFactory( - Table table, - Schema schema, - RowType flinkSchema, - Map props, - PartitionSpec spec, - int[] equalityFieldIds, - Schema eqDeleteRowSchema) { - this(table, schema, flinkSchema, props, spec, equalityFieldIds, eqDeleteRowSchema, null); - } - - /** - * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.12.0. - * Position deletes that include row data are no longer supported. Use {@link - * #FlinkAppenderFactory(Table, Schema, RowType, Map, PartitionSpec, int[], Schema)} instead. - */ - @Deprecated - public FlinkAppenderFactory( - Table table, - Schema schema, - RowType flinkSchema, - Map props, - PartitionSpec spec, - int[] equalityFieldIds, - Schema eqDeleteRowSchema, - Schema posDeleteRowSchema) { - Preconditions.checkNotNull(table, "Table shouldn't be null"); - this.table = table; - this.schema = schema; - this.flinkSchema = flinkSchema; - this.props = props; - this.spec = spec; - this.equalityFieldIds = equalityFieldIds; - this.eqDeleteRowSchema = eqDeleteRowSchema; - this.posDeleteRowSchema = posDeleteRowSchema; - } - - private RowType lazyEqDeleteFlinkSchema() { - if (eqDeleteFlinkSchema == null) { - Preconditions.checkNotNull(eqDeleteRowSchema, "Equality delete row schema shouldn't be null"); - this.eqDeleteFlinkSchema = FlinkSchemaUtil.convert(eqDeleteRowSchema); - } - return eqDeleteFlinkSchema; - } - - private RowType lazyPosDeleteFlinkSchema() { - if (posDeleteFlinkSchema == null) { - Preconditions.checkNotNull(posDeleteRowSchema, "Pos-delete row schema shouldn't be null"); - this.posDeleteFlinkSchema = FlinkSchemaUtil.convert(posDeleteRowSchema); - } - return this.posDeleteFlinkSchema; - } - - @Override - public FileAppender newAppender(OutputFile outputFile, FileFormat format) { - MetricsConfig metricsConfig = MetricsConfig.forTable(table); - try { - switch (format) { - case AVRO: - return Avro.write(outputFile) - .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema)) - .setAll(props) - .schema(schema) - .metricsConfig(metricsConfig) - .overwrite() - .build(); - - case ORC: - return ORC.write(outputFile) - .createWriterFunc( - (iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema)) - .setAll(props) - .metricsConfig(metricsConfig) - .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 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 newEqDeleteWriter( - EncryptedOutputFile outputFile, FileFormat format, StructLike partition) { - Preconditions.checkState( - equalityFieldIds != null && equalityFieldIds.length > 0, - "Equality field ids shouldn't be null or empty when creating equality-delete writer"); - Preconditions.checkNotNull( - eqDeleteRowSchema, - "Equality delete row schema shouldn't be null when creating equality-delete writer"); - - MetricsConfig metricsConfig = MetricsConfig.forTable(table); - try { - switch (format) { - case AVRO: - return Avro.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc(ignore -> new FlinkAvroWriter(lazyEqDeleteFlinkSchema())) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(eqDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .equalityFieldIds(equalityFieldIds) - .buildEqualityWriter(); - - case ORC: - return ORC.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - (iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(eqDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .equalityFieldIds(equalityFieldIds) - .buildEqualityWriter(); - - case PARQUET: - return Parquet.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - msgType -> FlinkParquetWriters.buildWriter(lazyEqDeleteFlinkSchema(), msgType)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(eqDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .equalityFieldIds(equalityFieldIds) - .buildEqualityWriter(); - - default: - throw new UnsupportedOperationException( - "Cannot write equality-deletes for unsupported file format: " + format); - } - } catch (IOException e) { - throw new UncheckedIOException(e); - } - } - - @Override - public PositionDeleteWriter newPosDeleteWriter( - EncryptedOutputFile outputFile, FileFormat format, StructLike partition) { - MetricsConfig metricsConfig = MetricsConfig.forPositionDelete(); - try { - switch (format) { - case AVRO: - return Avro.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc(ignore -> new FlinkAvroWriter(lazyPosDeleteFlinkSchema())) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(posDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .buildPositionWriter(); - - case ORC: - RowType orcPosDeleteSchema = - FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema)); - return ORC.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - (iSchema, typDesc) -> FlinkOrcWriter.buildWriter(orcPosDeleteSchema, iSchema)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(posDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .transformPaths(path -> StringData.fromString(path.toString())) - .buildPositionWriter(); - - case PARQUET: - RowType flinkPosDeleteSchema = - FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema)); - return Parquet.writeDeletes(outputFile.encryptingOutputFile()) - .createWriterFunc( - msgType -> FlinkParquetWriters.buildWriter(flinkPosDeleteSchema, msgType)) - .withPartition(partition) - .overwrite() - .setAll(props) - .metricsConfig(metricsConfig) - .rowSchema(posDeleteRowSchema) - .withSpec(spec) - .withKeyMetadata(outputFile.keyMetadata()) - .transformPaths(path -> StringData.fromString(path.toString())) - .buildPositionWriter(); - - default: - throw new UnsupportedOperationException( - "Cannot write pos-deletes for unsupported file format: " + format); - } - } catch (IOException e) { - throw new UncheckedIOException(e); - } - } -} diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java deleted file mode 100644 index dd89f43483b0..000000000000 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * 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.util.List; -import org.apache.flink.table.data.RowData; -import org.apache.flink.table.types.logical.RowType; -import org.apache.iceberg.Schema; -import org.apache.iceberg.flink.FlinkSchemaUtil; -import org.apache.iceberg.flink.RowDataWrapper; -import org.apache.iceberg.flink.SimpleDataUtil; -import org.apache.iceberg.io.FileAppenderFactory; -import org.apache.iceberg.io.TestAppenderFactory; -import org.apache.iceberg.util.ArrayUtil; -import org.apache.iceberg.util.StructLikeSet; - -public class TestFlinkAppenderFactory extends TestAppenderFactory { - - private final RowType rowType = FlinkSchemaUtil.convert(SCHEMA); - - @Override - protected FileAppenderFactory createAppenderFactory( - List equalityFieldIds, Schema eqDeleteSchema, Schema posDeleteRowSchema) { - return new FlinkAppenderFactory( - table, - table.schema(), - rowType, - table.properties(), - table.spec(), - ArrayUtil.toIntArray(equalityFieldIds), - eqDeleteSchema, - posDeleteRowSchema); - } - - @Override - protected RowData createRow(Integer id, String data) { - return SimpleDataUtil.createRowData(id, data); - } - - @Override - protected StructLikeSet expectedRowSet(Iterable rows) { - StructLikeSet set = StructLikeSet.create(table.schema().asStruct()); - for (RowData row : rows) { - RowDataWrapper wrapper = new RowDataWrapper(rowType, table.schema().asStruct()); - set.add(wrapper.wrap(row)); - } - return set; - } -} diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java index 5f13b8aac45b..b97bf56cd844 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java @@ -64,7 +64,7 @@ class SparkFileWriterFactory extends RegistryBasedFileWriterFactory writeProperties; /** - * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. Use {@link * #SparkFileWriterFactory(Table, FileFormat, Schema, StructType, SortOrder, FileFormat, * int[], Schema, StructType, SortOrder, Map)} instead. @@ -164,7 +164,7 @@ public PositionDeleteWriter newPositionDeleteWriter( if (!useDeprecatedPositionDeleteWriter) { return super.newPositionDeleteWriter(file, spec, partition); } else { - LOG.warn("Position deletes with deleted rows are deprecated and will be removed in 1.12.0."); + LOG.warn("Position deletes with deleted rows are deprecated and will be removed in 1.13.0."); Map properties = table == null ? ImmutableMap.of() : table.properties(); MetricsConfig metricsConfig = MetricsConfig.forPositionDelete(); @@ -301,7 +301,7 @@ Builder equalityDeleteSortOrder(SortOrder newEqualityDeleteSortOrder) { } /** - * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ @Deprecated @@ -311,7 +311,7 @@ Builder positionDeleteRowSchema(Schema newPositionDeleteRowSchema) { } /** - * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ @Deprecated diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java index f926bd96389a..20f57a6908f6 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java @@ -833,7 +833,7 @@ long targetDataFileSize() { return targetDataFileSize; } - /* @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + /* @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ StructType deleteSparkType() { diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java index 5f13b8aac45b..b97bf56cd844 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java @@ -64,7 +64,7 @@ class SparkFileWriterFactory extends RegistryBasedFileWriterFactory writeProperties; /** - * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. Use {@link * #SparkFileWriterFactory(Table, FileFormat, Schema, StructType, SortOrder, FileFormat, * int[], Schema, StructType, SortOrder, Map)} instead. @@ -164,7 +164,7 @@ public PositionDeleteWriter newPositionDeleteWriter( if (!useDeprecatedPositionDeleteWriter) { return super.newPositionDeleteWriter(file, spec, partition); } else { - LOG.warn("Position deletes with deleted rows are deprecated and will be removed in 1.12.0."); + LOG.warn("Position deletes with deleted rows are deprecated and will be removed in 1.13.0."); Map properties = table == null ? ImmutableMap.of() : table.properties(); MetricsConfig metricsConfig = MetricsConfig.forPositionDelete(); @@ -301,7 +301,7 @@ Builder equalityDeleteSortOrder(SortOrder newEqualityDeleteSortOrder) { } /** - * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ @Deprecated @@ -311,7 +311,7 @@ Builder positionDeleteRowSchema(Schema newPositionDeleteRowSchema) { } /** - * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ @Deprecated diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java index f0a58fc42107..8de62f1169e2 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java @@ -911,7 +911,7 @@ long targetDataFileSize() { return targetDataFileSize; } - /* @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + /* @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ StructType deleteSparkType() { diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java index 5f13b8aac45b..b97bf56cd844 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java @@ -64,7 +64,7 @@ class SparkFileWriterFactory extends RegistryBasedFileWriterFactory writeProperties; /** - * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. Use {@link * #SparkFileWriterFactory(Table, FileFormat, Schema, StructType, SortOrder, FileFormat, * int[], Schema, StructType, SortOrder, Map)} instead. @@ -164,7 +164,7 @@ public PositionDeleteWriter newPositionDeleteWriter( if (!useDeprecatedPositionDeleteWriter) { return super.newPositionDeleteWriter(file, spec, partition); } else { - LOG.warn("Position deletes with deleted rows are deprecated and will be removed in 1.12.0."); + LOG.warn("Position deletes with deleted rows are deprecated and will be removed in 1.13.0."); Map properties = table == null ? ImmutableMap.of() : table.properties(); MetricsConfig metricsConfig = MetricsConfig.forPositionDelete(); @@ -301,7 +301,7 @@ Builder equalityDeleteSortOrder(SortOrder newEqualityDeleteSortOrder) { } /** - * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ @Deprecated @@ -311,7 +311,7 @@ Builder positionDeleteRowSchema(Schema newPositionDeleteRowSchema) { } /** - * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ @Deprecated diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java index 89f6e1c350fc..a4ae338f1db6 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java @@ -914,7 +914,7 @@ long targetDataFileSize() { return targetDataFileSize; } - /* @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0. + /* @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.13.0. * Position deletes that include row data are no longer supported. */ StructType deleteSparkType() {