From e8f96e2bd86d4cad25ae11ab62c0ea533866dc66 Mon Sep 17 00:00:00 2001 From: Hongyue Zhang Date: Tue, 25 Aug 2026 10:43:40 -0700 Subject: [PATCH] Core, Data, Spark, Flink: Defer removal of position deletes with row data to 1.13.0 The removal of the ability to write position deletes with row data (PDWR) was scheduled for 1.12.0 (see #17706). The 1.12.0 release candidate is being cut now and the removal touches public API across Core, Data and Spark 3.5/4.0/4.1, so it should not be rushed through review. This moves the removal target of those deprecations from 1.12.0 to 1.13.0. The deprecations themselves are unchanged: everything is still deprecated as of 1.11.0 and callers should still migrate off the row-carrying overloads. Only the stated removal release moves out by one cycle. This mirrors #14392, which retargeted a batch of removal versions in the other direction. Markers moved to 1.13.0: - PositionDelete.set(CharSequence, long, R) and row() - PositionDeleteWriter constructor (appender type narrowing) - RewriteTablePathUtil.PositionDeleteReaderWriter.writer(..., Schema) - GenericFileWriterFactory positionDeleteRowSchema constructors, the builder setter, and the nine configureDataWrite/configureEqualityDelete/ configurePositionDelete methods superseded by FormatModelRegistry, whose removal was bundled into the same change - SparkFileWriterFactory positionDeleteRowSchema/positionDeleteSparkType builder setters, the row-schema constructor, and the runtime warning message - SparkPositionDeltaWrite.Context.deleteSparkType() Additionally, PositionDeltaWriter.delete(CharSequence, long, T, PartitionSpec, StructLike) is now deprecated for removal in 1.13.0. It was previously not deprecated at all, so without this its deprecation cycle would only start in 1.13.0 and the row parameter could not be dropped until 1.14.0. Deprecating it here keeps that narrowing on schedule for 1.13.0 alongside the rest. The annotation is additive: the overload stays abstract, so implementations are unaffected, while callers can migrate to the four-argument overload today. All three SparkPositionDeltaWrite call sites already use it; the only override is BasePositionDeltaWriter, which already calls the deprecated PositionDelete.set(CharSequence, long, R). Flink is not deferred. Its only PDWR surface was FlinkAppenderFactory, whose entire position delete path is built on row data. That class was deprecated as of 1.11.0 in favor of FlinkFileWriterFactory for removal in 1.12.0, and nothing in the repo references it outside its own test, so it is removed here on schedule from all three Flink versions rather than carried into 1.13.0. That also drops org.apache.iceberg.io.TestAppenderFactory from iceberg-data: the three TestFlinkAppenderFactory classes were its only subclasses, so with those gone the abstract base exercises nothing. iceberg-flink is not a revapi project and TestAppenderFactory is a test class, so neither removal needs a revapi baseline entry. The REST scan-planning markers in BaseScanTaskResponse, PlanTableScanRequest and PlanTableScanResponse are left alone; they are handled separately in #17638. --- .../apache/iceberg/RewriteTablePathUtil.java | 2 +- .../iceberg/deletes/PositionDelete.java | 4 +- .../iceberg/deletes/PositionDeleteWriter.java | 2 +- .../iceberg/io/PositionDeltaWriter.java | 4 + .../data/GenericFileWriterFactory.java | 24 +- .../iceberg/io/TestAppenderFactory.java | 355 ------------------ .../flink/sink/FlinkAppenderFactory.java | 296 --------------- .../flink/sink/TestFlinkAppenderFactory.java | 65 ---- .../flink/sink/FlinkAppenderFactory.java | 296 --------------- .../flink/sink/TestFlinkAppenderFactory.java | 65 ---- .../flink/sink/FlinkAppenderFactory.java | 296 --------------- .../flink/sink/TestFlinkAppenderFactory.java | 65 ---- .../spark/source/SparkFileWriterFactory.java | 8 +- .../spark/source/SparkPositionDeltaWrite.java | 2 +- .../spark/source/SparkFileWriterFactory.java | 8 +- .../spark/source/SparkPositionDeltaWrite.java | 2 +- .../spark/source/SparkFileWriterFactory.java | 8 +- .../spark/source/SparkPositionDeltaWrite.java | 2 +- 18 files changed, 35 insertions(+), 1469 deletions(-) delete mode 100644 data/src/test/java/org/apache/iceberg/io/TestAppenderFactory.java delete mode 100644 flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java delete mode 100644 flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java delete mode 100644 flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java delete mode 100644 flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java delete mode 100644 flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java delete mode 100644 flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkAppenderFactory.java 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() {