You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2024/12/10 19:36:17 UTC
(iceberg) branch main updated: Core, Flink, Spark: Drop deprecated APIs scheduled for removal in 1.8.0 (#11721)
This is an automated email from the ASF dual-hosted git repository.
etudenhoefner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 28e81809e6 Core, Flink, Spark: Drop deprecated APIs scheduled for removal in 1.8.0 (#11721)
28e81809e6 is described below
commit 28e81809e64b3d6c5b1f8d40f53fcb5edc1ec3e9
Author: Piotr Findeisen <pi...@gmail.com>
AuthorDate: Tue Dec 10 20:36:12 2024 +0100
Core, Flink, Spark: Drop deprecated APIs scheduled for removal in 1.8.0 (#11721)
---
.palantir/revapi.yml | 26 +++
.../parquet/VectorizedColumnIterator.java | 26 ---
...orizedDictionaryEncodedParquetValuesReader.java | 21 ---
.../vectorized/parquet/VectorizedPageIterator.java | 42 -----
.../VectorizedParquetDefinitionLevelReader.java | 51 ------
.../java/org/apache/iceberg/deletes/Deletes.java | 78 +--------
.../apache/iceberg/deletes/TestPositionFilter.java | 32 +---
.../apache/iceberg/flink/data/FlinkAvroReader.java | 181 ---------------------
.../data/TestFlinkAvroDeprecatedReaderWriter.java | 38 -----
.../iceberg/flink/data/TestRowProjection.java | 69 ++------
.../apache/iceberg/flink/data/FlinkAvroReader.java | 181 ---------------------
.../data/TestFlinkAvroDeprecatedReaderWriter.java | 38 -----
.../iceberg/flink/data/TestRowProjection.java | 67 ++------
.../apache/iceberg/flink/data/FlinkAvroReader.java | 181 ---------------------
.../data/TestFlinkAvroDeprecatedReaderWriter.java | 38 -----
.../iceberg/flink/data/TestRowProjection.java | 68 ++------
.../apache/iceberg/spark/data/SparkAvroReader.java | 180 --------------------
17 files changed, 85 insertions(+), 1232 deletions(-)
diff --git a/.palantir/revapi.yml b/.palantir/revapi.yml
index fade79326a..18c63fbe7b 100644
--- a/.palantir/revapi.yml
+++ b/.palantir/revapi.yml
@@ -1145,6 +1145,32 @@ acceptedBreaks:
new: "method org.apache.iceberg.BaseMetastoreOperations.CommitStatus org.apache.iceberg.BaseMetastoreTableOperations::checkCommitStatus(java.lang.String,\
\ org.apache.iceberg.TableMetadata)"
justification: "Removing deprecated code"
+ "1.7.0":
+ org.apache.iceberg:iceberg-core:
+ - code: "java.method.removed"
+ old: "method <T extends org.apache.iceberg.StructLike> org.apache.iceberg.deletes.PositionDeleteIndex\
+ \ org.apache.iceberg.deletes.Deletes::toPositionIndex(java.lang.CharSequence,\
+ \ java.util.List<org.apache.iceberg.io.CloseableIterable<T>>)"
+ justification: "Removing deprecated code"
+ - code: "java.method.removed"
+ old: "method <T extends org.apache.iceberg.StructLike> org.apache.iceberg.deletes.PositionDeleteIndex\
+ \ org.apache.iceberg.deletes.Deletes::toPositionIndex(java.lang.CharSequence,\
+ \ java.util.List<org.apache.iceberg.io.CloseableIterable<T>>, java.util.concurrent.ExecutorService)"
+ justification: "Removing deprecated code"
+ - code: "java.method.removed"
+ old: "method <T> org.apache.iceberg.io.CloseableIterable<T> org.apache.iceberg.deletes.Deletes::streamingFilter(org.apache.iceberg.io.CloseableIterable<T>,\
+ \ java.util.function.Function<T, java.lang.Long>, org.apache.iceberg.io.CloseableIterable<java.lang.Long>)"
+ justification: "Removing deprecated code"
+ - code: "java.method.removed"
+ old: "method <T> org.apache.iceberg.io.CloseableIterable<T> org.apache.iceberg.deletes.Deletes::streamingFilter(org.apache.iceberg.io.CloseableIterable<T>,\
+ \ java.util.function.Function<T, java.lang.Long>, org.apache.iceberg.io.CloseableIterable<java.lang.Long>,\
+ \ org.apache.iceberg.deletes.DeleteCounter)"
+ justification: "Removing deprecated code"
+ - code: "java.method.removed"
+ old: "method <T> org.apache.iceberg.io.CloseableIterable<T> org.apache.iceberg.deletes.Deletes::streamingMarker(org.apache.iceberg.io.CloseableIterable<T>,\
+ \ java.util.function.Function<T, java.lang.Long>, org.apache.iceberg.io.CloseableIterable<java.lang.Long>,\
+ \ java.util.function.Consumer<T>)"
+ justification: "Removing deprecated code"
apache-iceberg-0.14.0:
org.apache.iceberg:iceberg-api:
- code: "java.class.defaultSerializationChanged"
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedColumnIterator.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedColumnIterator.java
index 54445e4245..8e52fd1a28 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedColumnIterator.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedColumnIterator.java
@@ -214,24 +214,6 @@ public class VectorizedColumnIterator extends BaseColumnIterator {
}
}
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- public class FixedWidthTypeBinaryBatchReader extends BatchReader {
- @Override
- protected int nextBatchOf(
- final FieldVector vector,
- final int expectedBatchSize,
- final int numValsInVector,
- final int typeWidth,
- NullabilityHolder holder) {
- return vectorizedPageIterator
- .fixedWidthBinaryPageReader()
- .nextBatch(vector, expectedBatchSize, numValsInVector, typeWidth, holder);
- }
- }
-
public class BooleanBatchReader extends BatchReader {
@Override
protected int nextBatchOf(
@@ -282,14 +264,6 @@ public class VectorizedColumnIterator extends BaseColumnIterator {
return new VarWidthTypeBatchReader();
}
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- public FixedWidthTypeBinaryBatchReader fixedWidthTypeBinaryBatchReader() {
- return new FixedWidthTypeBinaryBatchReader();
- }
-
public BooleanBatchReader booleanBatchReader() {
return new BooleanBatchReader();
}
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedDictionaryEncodedParquetValuesReader.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedDictionaryEncodedParquetValuesReader.java
index 848bb2d936..1a00137fad 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedDictionaryEncodedParquetValuesReader.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedDictionaryEncodedParquetValuesReader.java
@@ -137,19 +137,6 @@ public class VectorizedDictionaryEncodedParquetValuesReader
}
}
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- class FixedWidthBinaryDictEncodedReader extends BaseDictEncodedReader {
- @Override
- protected void nextVal(
- FieldVector vector, Dictionary dict, int idx, int currentVal, int typeWidth) {
- ByteBuffer buffer = dict.decodeToBinary(currentVal).toByteBuffer();
- vector.getDataBuffer().setBytes((long) idx * typeWidth, buffer);
- }
- }
-
class VarWidthBinaryDictEncodedReader extends BaseDictEncodedReader {
@Override
protected void nextVal(
@@ -203,14 +190,6 @@ public class VectorizedDictionaryEncodedParquetValuesReader
return new DoubleDictEncodedReader();
}
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- public FixedWidthBinaryDictEncodedReader fixedWidthBinaryDictEncodedReader() {
- return new FixedWidthBinaryDictEncodedReader();
- }
-
public VarWidthBinaryDictEncodedReader varWidthBinaryDictEncodedReader() {
return new VarWidthBinaryDictEncodedReader();
}
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedPageIterator.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedPageIterator.java
index d36521273d..b97eb15455 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedPageIterator.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedPageIterator.java
@@ -21,7 +21,6 @@ package org.apache.iceberg.arrow.vectorized.parquet;
import java.io.IOException;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.IntVector;
-import org.apache.arrow.vector.VarBinaryVector;
import org.apache.iceberg.arrow.vectorized.NullabilityHolder;
import org.apache.iceberg.parquet.BasePageIterator;
import org.apache.iceberg.parquet.ParquetUtil;
@@ -417,39 +416,6 @@ public class VectorizedPageIterator extends BasePageIterator {
}
}
- /**
- * Method for reading batches of fixed width binary type (e.g. BYTE[7]). Spark does not support
- * fixed width binary data type. To work around this limitation, the data is read as fixed width
- * binary from parquet and stored in a {@link VarBinaryVector} in Arrow.
- *
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- class FixedWidthBinaryPageReader extends BasePageReader {
- @Override
- protected void nextVal(
- FieldVector vector, int batchSize, int numVals, int typeWidth, NullabilityHolder holder) {
- vectorizedDefinitionLevelReader
- .fixedWidthBinaryReader()
- .nextBatch(vector, numVals, typeWidth, batchSize, holder, plainValuesReader);
- }
-
- @Override
- protected void nextDictEncodedVal(
- FieldVector vector, int batchSize, int numVals, int typeWidth, NullabilityHolder holder) {
- vectorizedDefinitionLevelReader
- .fixedWidthBinaryReader()
- .nextDictEncodedBatch(
- vector,
- numVals,
- typeWidth,
- batchSize,
- holder,
- dictionaryEncodedValuesReader,
- dictionary);
- }
- }
-
/** Method for reading batches of booleans. */
class BooleanPageReader extends BasePageReader {
@Override
@@ -499,14 +465,6 @@ public class VectorizedPageIterator extends BasePageIterator {
return new VarWidthTypePageReader();
}
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- FixedWidthBinaryPageReader fixedWidthBinaryPageReader() {
- return new FixedWidthBinaryPageReader();
- }
-
BooleanPageReader booleanPageReader() {
return new BooleanPageReader();
}
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedParquetDefinitionLevelReader.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedParquetDefinitionLevelReader.java
index e8ec7bee8f..4df648ee3d 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedParquetDefinitionLevelReader.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedParquetDefinitionLevelReader.java
@@ -27,7 +27,6 @@ import org.apache.arrow.vector.BitVectorHelper;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.FixedSizeBinaryVector;
import org.apache.arrow.vector.IntVector;
-import org.apache.arrow.vector.VarBinaryVector;
import org.apache.iceberg.arrow.vectorized.NullabilityHolder;
import org.apache.iceberg.parquet.ParquetUtil;
import org.apache.iceberg.parquet.ValuesAsBytesReader;
@@ -493,48 +492,6 @@ public final class VectorizedParquetDefinitionLevelReader
}
}
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- class FixedWidthBinaryReader extends BaseReader {
- @Override
- protected void nextVal(
- FieldVector vector,
- int idx,
- ValuesAsBytesReader valuesReader,
- int typeWidth,
- byte[] byteArray) {
- ByteBuffer buffer = valuesReader.getBuffer(typeWidth);
- ((VarBinaryVector) vector)
- .setSafe(
- idx,
- buffer.array(),
- buffer.position() + buffer.arrayOffset(),
- buffer.limit() - buffer.position());
- }
-
- @Override
- protected void nextDictEncodedVal(
- FieldVector vector,
- int idx,
- VectorizedDictionaryEncodedParquetValuesReader reader,
- int numValuesToRead,
- Dictionary dict,
- NullabilityHolder nullabilityHolder,
- int typeWidth,
- Mode mode) {
- if (Mode.RLE.equals(mode)) {
- reader
- .fixedWidthBinaryDictEncodedReader()
- .nextBatch(vector, idx, numValuesToRead, dict, nullabilityHolder, typeWidth);
- } else if (Mode.PACKED.equals(mode)) {
- ByteBuffer buffer = dict.decodeToBinary(reader.readInteger()).toByteBuffer();
- vector.getDataBuffer().setBytes((long) idx * typeWidth, buffer);
- }
- }
- }
-
class FixedSizeBinaryReader extends BaseReader {
@Override
protected void nextVal(
@@ -736,14 +693,6 @@ public final class VectorizedParquetDefinitionLevelReader
return new TimestampInt96Reader();
}
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- FixedWidthBinaryReader fixedWidthBinaryReader() {
- return new FixedWidthBinaryReader();
- }
-
FixedSizeBinaryReader fixedSizeBinaryReader() {
return new FixedSizeBinaryReader();
}
diff --git a/core/src/main/java/org/apache/iceberg/deletes/Deletes.java b/core/src/main/java/org/apache/iceberg/deletes/Deletes.java
index 0838e9c0dd..46df91982a 100644
--- a/core/src/main/java/org/apache/iceberg/deletes/Deletes.java
+++ b/core/src/main/java/org/apache/iceberg/deletes/Deletes.java
@@ -21,7 +21,6 @@ package org.apache.iceberg.deletes;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.List;
-import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
@@ -38,10 +37,8 @@ import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.CharSequenceMap;
import org.apache.iceberg.util.Filter;
-import org.apache.iceberg.util.ParallelIterable;
import org.apache.iceberg.util.SortedMerge;
import org.apache.iceberg.util.StructLikeSet;
-import org.apache.iceberg.util.ThreadPools;
public class Deletes {
@@ -132,9 +129,8 @@ public class Deletes {
/**
* Builds a map of position delete indexes by path.
*
- * <p>Unlike {@link #toPositionIndex(CharSequence, List)}, this method builds a position delete
- * index for each referenced data file and does not filter deletes. This can be useful when the
- * entire delete file content is needed (e.g. caching).
+ * <p>This method builds a position delete index for each referenced data file and does not filter
+ * deletes. This can be useful when the entire delete file content is needed (e.g. caching).
*
* @param posDeletes position deletes
* @param file the source delete file for the deletes
@@ -173,37 +169,6 @@ public class Deletes {
return CloseableIterable.transform(filteredRows, row -> (Long) POSITION_ACCESSOR.get(row));
}
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0; use delete loaders.
- */
- @Deprecated
- public static <T extends StructLike> PositionDeleteIndex toPositionIndex(
- CharSequence dataLocation, List<CloseableIterable<T>> deleteFiles) {
- return toPositionIndex(dataLocation, deleteFiles, ThreadPools.getDeleteWorkerPool());
- }
-
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0; use delete loaders.
- */
- @Deprecated
- public static <T extends StructLike> PositionDeleteIndex toPositionIndex(
- CharSequence dataLocation,
- List<CloseableIterable<T>> deleteFiles,
- ExecutorService deleteWorkerPool) {
- DataFileFilter<T> locationFilter = new DataFileFilter<>(dataLocation);
- List<CloseableIterable<Long>> positions =
- Lists.transform(
- deleteFiles,
- deletes ->
- CloseableIterable.transform(
- locationFilter.filter(deletes), row -> (Long) POSITION_ACCESSOR.get(row)));
- if (positions.size() > 1 && deleteWorkerPool != null) {
- return toPositionIndex(new ParallelIterable<>(positions, deleteWorkerPool));
- } else {
- return toPositionIndex(CloseableIterable.concat(positions));
- }
- }
-
public static PositionDeleteIndex toPositionIndex(CloseableIterable<Long> posDeletes) {
return toPositionIndex(posDeletes, ImmutableList.of());
}
@@ -219,45 +184,6 @@ public class Deletes {
}
}
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- public static <T> CloseableIterable<T> streamingFilter(
- CloseableIterable<T> rows,
- Function<T, Long> rowToPosition,
- CloseableIterable<Long> posDeletes) {
- return streamingFilter(rows, rowToPosition, posDeletes, new DeleteCounter());
- }
-
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- public static <T> CloseableIterable<T> streamingFilter(
- CloseableIterable<T> rows,
- Function<T, Long> rowToPosition,
- CloseableIterable<Long> posDeletes,
- DeleteCounter counter) {
- PositionDeleteIndex positionIndex = toPositionIndex(posDeletes);
- Predicate<T> isDeleted = row -> positionIndex.isDeleted(rowToPosition.apply(row));
- return filterDeleted(rows, isDeleted, counter);
- }
-
- /**
- * @deprecated since 1.7.0, will be removed in 1.8.0.
- */
- @Deprecated
- public static <T> CloseableIterable<T> streamingMarker(
- CloseableIterable<T> rows,
- Function<T, Long> rowToPosition,
- CloseableIterable<Long> posDeletes,
- Consumer<T> markRowDeleted) {
- PositionDeleteIndex positionIndex = toPositionIndex(posDeletes);
- Predicate<T> isDeleted = row -> positionIndex.isDeleted(rowToPosition.apply(row));
- return markDeleted(rows, isDeleted, markRowDeleted);
- }
-
public static CloseableIterable<Long> deletePositions(
CharSequence dataLocation, CloseableIterable<StructLike> deleteFile) {
return deletePositions(dataLocation, ImmutableList.of(deleteFile));
diff --git a/core/src/test/java/org/apache/iceberg/deletes/TestPositionFilter.java b/core/src/test/java/org/apache/iceberg/deletes/TestPositionFilter.java
index 2a629b2dc2..5def0c70bb 100644
--- a/core/src/test/java/org/apache/iceberg/deletes/TestPositionFilter.java
+++ b/core/src/test/java/org/apache/iceberg/deletes/TestPositionFilter.java
@@ -21,11 +21,7 @@ package org.apache.iceberg.deletes;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Predicate;
-import java.util.stream.Stream;
import org.apache.avro.util.Utf8;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.TestHelpers.Row;
@@ -33,10 +29,7 @@ import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.MethodSource;
public class TestPositionFilter {
@Test
@@ -134,16 +127,8 @@ public class TestPositionFilter {
.containsExactlyElementsOf(Lists.newArrayList(1L, 2L, 5L, 6L, 8L));
}
- static Stream<ExecutorService> executorServiceProvider() {
- return Stream.of(
- null,
- MoreExecutors.getExitingExecutorService(
- (ThreadPoolExecutor) Executors.newFixedThreadPool(4)));
- }
-
- @ParameterizedTest
- @MethodSource("executorServiceProvider")
- public void testCombinedPositionSetRowFilter(ExecutorService executorService) {
+ @Test
+ public void testCombinedPositionSetRowFilter() {
CloseableIterable<StructLike> positionDeletes1 =
CloseableIterable.withNoopClose(
Lists.newArrayList(
@@ -175,13 +160,14 @@ public class TestPositionFilter {
Row.of(8L, "i"),
Row.of(9L, "j")));
+ CloseableIterable<Long> positions =
+ CloseableIterable.transform(
+ CloseableIterable.filter(
+ CloseableIterable.concat(ImmutableList.of(positionDeletes1, positionDeletes2)),
+ row -> "file_a.avro".equals(row.get(0, String.class))),
+ row -> row.get(1, Long.class));
Predicate<StructLike> isDeleted =
- row ->
- Deletes.toPositionIndex(
- "file_a.avro",
- ImmutableList.of(positionDeletes1, positionDeletes2),
- executorService)
- .isDeleted(row.get(0, Long.class));
+ row -> Deletes.toPositionIndex(positions).isDeleted(row.get(0, Long.class));
CloseableIterable<StructLike> actual = CloseableIterable.filter(rows, isDeleted.negate());
assertThat(Iterables.transform(actual, row -> row.get(0, Long.class)))
diff --git a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroReader.java b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroReader.java
deleted file mode 100644
index baae91dd18..0000000000
--- a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroReader.java
+++ /dev/null
@@ -1,181 +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.data;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.function.Supplier;
-import org.apache.avro.LogicalType;
-import org.apache.avro.LogicalTypes;
-import org.apache.avro.Schema;
-import org.apache.avro.io.DatumReader;
-import org.apache.avro.io.Decoder;
-import org.apache.flink.table.data.RowData;
-import org.apache.iceberg.avro.AvroSchemaWithTypeVisitor;
-import org.apache.iceberg.avro.SupportsRowPosition;
-import org.apache.iceberg.avro.ValueReader;
-import org.apache.iceberg.avro.ValueReaders;
-import org.apache.iceberg.data.avro.DecoderResolver;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.iceberg.types.Type;
-import org.apache.iceberg.types.Types;
-
-/**
- * @deprecated will be removed in 1.8.0; use FlinkPlannedAvroReader instead.
- */
-@Deprecated
-public class FlinkAvroReader implements DatumReader<RowData>, SupportsRowPosition {
-
- private final Schema readSchema;
- private final ValueReader<RowData> reader;
- private Schema fileSchema = null;
-
- /**
- * @deprecated will be removed in 1.8.0; use FlinkPlannedAvroReader instead.
- */
- @Deprecated
- public FlinkAvroReader(org.apache.iceberg.Schema expectedSchema, Schema readSchema) {
- this(expectedSchema, readSchema, ImmutableMap.of());
- }
-
- /**
- * @deprecated will be removed in 1.8.0; use FlinkPlannedAvroReader instead.
- */
- @Deprecated
- @SuppressWarnings("unchecked")
- public FlinkAvroReader(
- org.apache.iceberg.Schema expectedSchema, Schema readSchema, Map<Integer, ?> constants) {
- this.readSchema = readSchema;
- this.reader =
- (ValueReader<RowData>)
- AvroSchemaWithTypeVisitor.visit(expectedSchema, readSchema, new ReadBuilder(constants));
- }
-
- @Override
- public void setSchema(Schema newFileSchema) {
- this.fileSchema = Schema.applyAliases(newFileSchema, readSchema);
- }
-
- @Override
- public RowData read(RowData reuse, Decoder decoder) throws IOException {
- return DecoderResolver.resolveAndRead(decoder, readSchema, fileSchema, reader, reuse);
- }
-
- @Override
- public void setRowPositionSupplier(Supplier<Long> posSupplier) {
- if (reader instanceof SupportsRowPosition) {
- ((SupportsRowPosition) reader).setRowPositionSupplier(posSupplier);
- }
- }
-
- private static class ReadBuilder extends AvroSchemaWithTypeVisitor<ValueReader<?>> {
- private final Map<Integer, ?> idToConstant;
-
- private ReadBuilder(Map<Integer, ?> idToConstant) {
- this.idToConstant = idToConstant;
- }
-
- @Override
- public ValueReader<?> record(
- Types.StructType expected, Schema record, List<String> names, List<ValueReader<?>> fields) {
- return FlinkValueReaders.struct(fields, expected.asStructType(), idToConstant);
- }
-
- @Override
- public ValueReader<?> union(Type expected, Schema union, List<ValueReader<?>> options) {
- return ValueReaders.union(options);
- }
-
- @Override
- public ValueReader<?> array(
- Types.ListType expected, Schema array, ValueReader<?> elementReader) {
- return FlinkValueReaders.array(elementReader);
- }
-
- @Override
- public ValueReader<?> map(
- Types.MapType expected, Schema map, ValueReader<?> keyReader, ValueReader<?> valueReader) {
- return FlinkValueReaders.arrayMap(keyReader, valueReader);
- }
-
- @Override
- public ValueReader<?> map(Types.MapType expected, Schema map, ValueReader<?> valueReader) {
- return FlinkValueReaders.map(FlinkValueReaders.strings(), valueReader);
- }
-
- @Override
- public ValueReader<?> primitive(Type.PrimitiveType expected, Schema primitive) {
- LogicalType logicalType = primitive.getLogicalType();
- if (logicalType != null) {
- switch (logicalType.getName()) {
- case "date":
- return ValueReaders.ints();
-
- case "time-micros":
- return FlinkValueReaders.timeMicros();
-
- case "timestamp-millis":
- return FlinkValueReaders.timestampMills();
-
- case "timestamp-micros":
- return FlinkValueReaders.timestampMicros();
-
- case "decimal":
- LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) logicalType;
- return FlinkValueReaders.decimal(
- ValueReaders.decimalBytesReader(primitive),
- decimal.getPrecision(),
- decimal.getScale());
-
- case "uuid":
- return FlinkValueReaders.uuids();
-
- default:
- throw new IllegalArgumentException("Unknown logical type: " + logicalType);
- }
- }
-
- switch (primitive.getType()) {
- case NULL:
- return ValueReaders.nulls();
- case BOOLEAN:
- return ValueReaders.booleans();
- case INT:
- return ValueReaders.ints();
- case LONG:
- return ValueReaders.longs();
- case FLOAT:
- return ValueReaders.floats();
- case DOUBLE:
- return ValueReaders.doubles();
- case STRING:
- return FlinkValueReaders.strings();
- case FIXED:
- return ValueReaders.fixed(primitive.getFixedSize());
- case BYTES:
- return ValueReaders.bytes();
- case ENUM:
- return FlinkValueReaders.enums(primitive.getEnumSymbols());
- default:
- throw new IllegalArgumentException("Unsupported type: " + primitive);
- }
- }
- }
-}
diff --git a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroDeprecatedReaderWriter.java b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroDeprecatedReaderWriter.java
deleted file mode 100644
index 03910f4fda..0000000000
--- a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroDeprecatedReaderWriter.java
+++ /dev/null
@@ -1,38 +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.data;
-
-import java.io.File;
-import org.apache.iceberg.Files;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.avro.Avro;
-
-/**
- * @deprecated should be removed in 1.8.0; along with FlinkAvroReader.
- */
-@Deprecated
-public class TestFlinkAvroDeprecatedReaderWriter extends AbstractTestFlinkAvroReaderWriter {
-
- @Override
- protected Avro.ReadBuilder createAvroReadBuilder(File recordsFile, Schema schema) {
- return Avro.read(Files.localInput(recordsFile))
- .project(schema)
- .createReaderFunc(FlinkAvroReader::new);
- }
-}
diff --git a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java
index 282a6055cb..a199898362 100644
--- a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java
+++ b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java
@@ -24,9 +24,6 @@ import static org.assertj.core.api.Assertions.withPrecision;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.GenericMapData;
@@ -34,34 +31,20 @@ import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.iceberg.Files;
-import org.apache.iceberg.Parameter;
-import org.apache.iceberg.ParameterizedTestExtension;
-import org.apache.iceberg.Parameters;
import org.apache.iceberg.Schema;
import org.apache.iceberg.avro.Avro;
import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
-import org.junit.jupiter.api.TestTemplate;
-import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-@ExtendWith(ParameterizedTestExtension.class)
public class TestRowProjection {
@TempDir private Path temp;
- @Parameter(index = 0)
- protected Boolean useAvroPlannedReader;
-
- @Parameters(name = "useAvroPlannedReader={0}")
- protected static List<Object[]> parameters() {
- return Arrays.asList(new Object[] {Boolean.FALSE}, new Object[] {Boolean.TRUE});
- }
-
private RowData writeAndRead(String desc, Schema writeSchema, Schema readSchema, RowData row)
throws IOException {
File file = File.createTempFile("junit", desc + ".avro", temp.toFile());
@@ -78,20 +61,13 @@ public class TestRowProjection {
Avro.ReadBuilder builder =
Avro.read(Files.localInput(file))
.project(readSchema)
- .createReaderFunc(FlinkAvroReader::new);
- if (useAvroPlannedReader) {
- builder =
- Avro.read(Files.localInput(file))
- .project(readSchema)
- .createResolvingReader(FlinkPlannedAvroReader::create);
- }
-
+ .createResolvingReader(FlinkPlannedAvroReader::create);
Iterable<RowData> records = builder.build();
return Iterables.getOnlyElement(records);
}
- @TestTemplate
+ @Test
public void testFullProjection() throws Exception {
Schema schema =
new Schema(
@@ -106,7 +82,7 @@ public class TestRowProjection {
assertThat(projected.getString(1)).asString().isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testSpecialCharacterProjection() throws Exception {
Schema schema =
new Schema(
@@ -126,7 +102,7 @@ public class TestRowProjection {
assertThat(projected.getString(0)).asString().isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testReorderedFullProjection() throws Exception {
Schema schema =
new Schema(
@@ -146,7 +122,7 @@ public class TestRowProjection {
assertThat(projected.getLong(1)).isEqualTo(34);
}
- @TestTemplate
+ @Test
public void testReorderedProjection() throws Exception {
Schema schema =
new Schema(
@@ -168,7 +144,7 @@ public class TestRowProjection {
assertThat(projected.isNullAt(2)).isTrue();
}
- @TestTemplate
+ @Test
public void testRenamedAddedField() throws Exception {
Schema schema =
new Schema(
@@ -198,7 +174,7 @@ public class TestRowProjection {
assertThat(projected.isNullAt(3)).as("Should contain empty value on new column 4").isTrue();
}
- @TestTemplate
+ @Test
public void testEmptyProjection() throws Exception {
Schema schema =
new Schema(
@@ -213,7 +189,7 @@ public class TestRowProjection {
assertThat(projected.getArity()).isEqualTo(0);
}
- @TestTemplate
+ @Test
public void testBasicProjection() throws Exception {
Schema writeSchema =
new Schema(
@@ -236,7 +212,7 @@ public class TestRowProjection {
assertThat(projected.getString(0)).asString().isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testRename() throws Exception {
Schema writeSchema =
new Schema(
@@ -259,7 +235,7 @@ public class TestRowProjection {
.isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testNestedStructProjection() throws Exception {
Schema writeSchema =
new Schema(
@@ -325,7 +301,7 @@ public class TestRowProjection {
.isEqualTo(-1.539054f, withPrecision(0.000001f));
}
- @TestTemplate
+ @Test
public void testMapProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -367,19 +343,7 @@ public class TestRowProjection {
assertThat(projected.getMap(0)).isEqualTo(properties);
}
- private Map<String, ?> toStringMap(Map<?, ?> map) {
- Map<String, Object> stringMap = Maps.newHashMap();
- for (Map.Entry<?, ?> entry : map.entrySet()) {
- if (entry.getValue() instanceof CharSequence) {
- stringMap.put(entry.getKey().toString(), entry.getValue().toString());
- } else {
- stringMap.put(entry.getKey().toString(), entry.getValue());
- }
- }
- return stringMap;
- }
-
- @TestTemplate
+ @Test
public void testMapOfStructsProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -479,7 +443,7 @@ public class TestRowProjection {
.isEqualTo(52.995143f, withPrecision(0.000001f));
}
- @TestTemplate
+ @Test
public void testListProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -508,8 +472,7 @@ public class TestRowProjection {
assertThat(projected.getArray(0)).isEqualTo(values);
}
- @TestTemplate
- @SuppressWarnings("unchecked")
+ @Test
public void testListOfStructsProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -585,7 +548,7 @@ public class TestRowProjection {
assertThat(projectedP2.isNullAt(0)).as("Should project null z").isTrue();
}
- @TestTemplate
+ @Test
public void testAddedFieldsWithRequiredChildren() throws Exception {
Schema schema = new Schema(Types.NestedField.required(1, "a", Types.LongType.get()));
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroReader.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroReader.java
deleted file mode 100644
index baae91dd18..0000000000
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroReader.java
+++ /dev/null
@@ -1,181 +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.data;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.function.Supplier;
-import org.apache.avro.LogicalType;
-import org.apache.avro.LogicalTypes;
-import org.apache.avro.Schema;
-import org.apache.avro.io.DatumReader;
-import org.apache.avro.io.Decoder;
-import org.apache.flink.table.data.RowData;
-import org.apache.iceberg.avro.AvroSchemaWithTypeVisitor;
-import org.apache.iceberg.avro.SupportsRowPosition;
-import org.apache.iceberg.avro.ValueReader;
-import org.apache.iceberg.avro.ValueReaders;
-import org.apache.iceberg.data.avro.DecoderResolver;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.iceberg.types.Type;
-import org.apache.iceberg.types.Types;
-
-/**
- * @deprecated will be removed in 1.8.0; use FlinkPlannedAvroReader instead.
- */
-@Deprecated
-public class FlinkAvroReader implements DatumReader<RowData>, SupportsRowPosition {
-
- private final Schema readSchema;
- private final ValueReader<RowData> reader;
- private Schema fileSchema = null;
-
- /**
- * @deprecated will be removed in 1.8.0; use FlinkPlannedAvroReader instead.
- */
- @Deprecated
- public FlinkAvroReader(org.apache.iceberg.Schema expectedSchema, Schema readSchema) {
- this(expectedSchema, readSchema, ImmutableMap.of());
- }
-
- /**
- * @deprecated will be removed in 1.8.0; use FlinkPlannedAvroReader instead.
- */
- @Deprecated
- @SuppressWarnings("unchecked")
- public FlinkAvroReader(
- org.apache.iceberg.Schema expectedSchema, Schema readSchema, Map<Integer, ?> constants) {
- this.readSchema = readSchema;
- this.reader =
- (ValueReader<RowData>)
- AvroSchemaWithTypeVisitor.visit(expectedSchema, readSchema, new ReadBuilder(constants));
- }
-
- @Override
- public void setSchema(Schema newFileSchema) {
- this.fileSchema = Schema.applyAliases(newFileSchema, readSchema);
- }
-
- @Override
- public RowData read(RowData reuse, Decoder decoder) throws IOException {
- return DecoderResolver.resolveAndRead(decoder, readSchema, fileSchema, reader, reuse);
- }
-
- @Override
- public void setRowPositionSupplier(Supplier<Long> posSupplier) {
- if (reader instanceof SupportsRowPosition) {
- ((SupportsRowPosition) reader).setRowPositionSupplier(posSupplier);
- }
- }
-
- private static class ReadBuilder extends AvroSchemaWithTypeVisitor<ValueReader<?>> {
- private final Map<Integer, ?> idToConstant;
-
- private ReadBuilder(Map<Integer, ?> idToConstant) {
- this.idToConstant = idToConstant;
- }
-
- @Override
- public ValueReader<?> record(
- Types.StructType expected, Schema record, List<String> names, List<ValueReader<?>> fields) {
- return FlinkValueReaders.struct(fields, expected.asStructType(), idToConstant);
- }
-
- @Override
- public ValueReader<?> union(Type expected, Schema union, List<ValueReader<?>> options) {
- return ValueReaders.union(options);
- }
-
- @Override
- public ValueReader<?> array(
- Types.ListType expected, Schema array, ValueReader<?> elementReader) {
- return FlinkValueReaders.array(elementReader);
- }
-
- @Override
- public ValueReader<?> map(
- Types.MapType expected, Schema map, ValueReader<?> keyReader, ValueReader<?> valueReader) {
- return FlinkValueReaders.arrayMap(keyReader, valueReader);
- }
-
- @Override
- public ValueReader<?> map(Types.MapType expected, Schema map, ValueReader<?> valueReader) {
- return FlinkValueReaders.map(FlinkValueReaders.strings(), valueReader);
- }
-
- @Override
- public ValueReader<?> primitive(Type.PrimitiveType expected, Schema primitive) {
- LogicalType logicalType = primitive.getLogicalType();
- if (logicalType != null) {
- switch (logicalType.getName()) {
- case "date":
- return ValueReaders.ints();
-
- case "time-micros":
- return FlinkValueReaders.timeMicros();
-
- case "timestamp-millis":
- return FlinkValueReaders.timestampMills();
-
- case "timestamp-micros":
- return FlinkValueReaders.timestampMicros();
-
- case "decimal":
- LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) logicalType;
- return FlinkValueReaders.decimal(
- ValueReaders.decimalBytesReader(primitive),
- decimal.getPrecision(),
- decimal.getScale());
-
- case "uuid":
- return FlinkValueReaders.uuids();
-
- default:
- throw new IllegalArgumentException("Unknown logical type: " + logicalType);
- }
- }
-
- switch (primitive.getType()) {
- case NULL:
- return ValueReaders.nulls();
- case BOOLEAN:
- return ValueReaders.booleans();
- case INT:
- return ValueReaders.ints();
- case LONG:
- return ValueReaders.longs();
- case FLOAT:
- return ValueReaders.floats();
- case DOUBLE:
- return ValueReaders.doubles();
- case STRING:
- return FlinkValueReaders.strings();
- case FIXED:
- return ValueReaders.fixed(primitive.getFixedSize());
- case BYTES:
- return ValueReaders.bytes();
- case ENUM:
- return FlinkValueReaders.enums(primitive.getEnumSymbols());
- default:
- throw new IllegalArgumentException("Unsupported type: " + primitive);
- }
- }
- }
-}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroDeprecatedReaderWriter.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroDeprecatedReaderWriter.java
deleted file mode 100644
index 03910f4fda..0000000000
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroDeprecatedReaderWriter.java
+++ /dev/null
@@ -1,38 +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.data;
-
-import java.io.File;
-import org.apache.iceberg.Files;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.avro.Avro;
-
-/**
- * @deprecated should be removed in 1.8.0; along with FlinkAvroReader.
- */
-@Deprecated
-public class TestFlinkAvroDeprecatedReaderWriter extends AbstractTestFlinkAvroReaderWriter {
-
- @Override
- protected Avro.ReadBuilder createAvroReadBuilder(File recordsFile, Schema schema) {
- return Avro.read(Files.localInput(recordsFile))
- .project(schema)
- .createReaderFunc(FlinkAvroReader::new);
- }
-}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java
index f76e4c4942..004cc82348 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java
@@ -24,9 +24,6 @@ import static org.assertj.core.api.Assertions.withPrecision;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.GenericMapData;
@@ -34,35 +31,21 @@ import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.iceberg.Files;
-import org.apache.iceberg.Parameter;
-import org.apache.iceberg.ParameterizedTestExtension;
-import org.apache.iceberg.Parameters;
import org.apache.iceberg.Schema;
import org.apache.iceberg.avro.Avro;
import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.types.Types;
-import org.junit.jupiter.api.TestTemplate;
-import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-@ExtendWith(ParameterizedTestExtension.class)
public class TestRowProjection {
@TempDir private Path temp;
- @Parameter(index = 0)
- protected Boolean useAvroPlannedReader;
-
- @Parameters(name = "useAvroPlannedReader={0}")
- protected static List<Object[]> parameters() {
- return Arrays.asList(new Object[] {Boolean.FALSE}, new Object[] {Boolean.TRUE});
- }
-
private RowData writeAndRead(String desc, Schema writeSchema, Schema readSchema, RowData row)
throws IOException {
File file = File.createTempFile("junit", desc + ".avro", temp.toFile());
@@ -79,20 +62,14 @@ public class TestRowProjection {
Avro.ReadBuilder builder =
Avro.read(Files.localInput(file))
.project(readSchema)
- .createReaderFunc(FlinkAvroReader::new);
- if (useAvroPlannedReader) {
- builder =
- Avro.read(Files.localInput(file))
- .project(readSchema)
- .createResolvingReader(FlinkPlannedAvroReader::create);
- }
+ .createResolvingReader(FlinkPlannedAvroReader::create);
Iterable<RowData> records = builder.build();
return Iterables.getOnlyElement(records);
}
- @TestTemplate
+ @Test
public void testFullProjection() throws Exception {
Schema schema =
new Schema(
@@ -107,7 +84,7 @@ public class TestRowProjection {
assertThat(projected.getString(1)).asString().isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testSpecialCharacterProjection() throws Exception {
Schema schema =
new Schema(
@@ -127,7 +104,7 @@ public class TestRowProjection {
assertThat(projected.getString(0)).asString().isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testReorderedFullProjection() throws Exception {
Schema schema =
new Schema(
@@ -147,7 +124,7 @@ public class TestRowProjection {
assertThat(projected.getLong(1)).isEqualTo(34);
}
- @TestTemplate
+ @Test
public void testReorderedProjection() throws Exception {
Schema schema =
new Schema(
@@ -169,7 +146,7 @@ public class TestRowProjection {
assertThat(projected.isNullAt(2)).isTrue();
}
- @TestTemplate
+ @Test
public void testRenamedAddedField() throws Exception {
Schema schema =
new Schema(
@@ -199,7 +176,7 @@ public class TestRowProjection {
assertThat(projected.isNullAt(3)).as("Should contain empty value on new column 4").isTrue();
}
- @TestTemplate
+ @Test
public void testEmptyProjection() throws Exception {
Schema schema =
new Schema(
@@ -214,7 +191,7 @@ public class TestRowProjection {
assertThat(projected.getArity()).isEqualTo(0);
}
- @TestTemplate
+ @Test
public void testBasicProjection() throws Exception {
Schema writeSchema =
new Schema(
@@ -238,7 +215,7 @@ public class TestRowProjection {
assertThat(projected.getString(0)).asString().isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testRename() throws Exception {
Schema writeSchema =
new Schema(
@@ -261,7 +238,7 @@ public class TestRowProjection {
.isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testNestedStructProjection() throws Exception {
Schema writeSchema =
new Schema(
@@ -327,7 +304,7 @@ public class TestRowProjection {
.isEqualTo(-1.539054f, withPrecision(0.000001f));
}
- @TestTemplate
+ @Test
public void testMapProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -369,19 +346,7 @@ public class TestRowProjection {
assertThat(projected.getMap(0)).isEqualTo(properties);
}
- private Map<String, ?> toStringMap(Map<?, ?> map) {
- Map<String, Object> stringMap = Maps.newHashMap();
- for (Map.Entry<?, ?> entry : map.entrySet()) {
- if (entry.getValue() instanceof CharSequence) {
- stringMap.put(entry.getKey().toString(), entry.getValue().toString());
- } else {
- stringMap.put(entry.getKey().toString(), entry.getValue());
- }
- }
- return stringMap;
- }
-
- @TestTemplate
+ @Test
public void testMapOfStructsProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -481,7 +446,7 @@ public class TestRowProjection {
.isEqualTo(52.995143f, withPrecision(0.000001f));
}
- @TestTemplate
+ @Test
public void testListProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -510,7 +475,7 @@ public class TestRowProjection {
assertThat(projected.getArray(0)).isEqualTo(values);
}
- @TestTemplate
+ @Test
@SuppressWarnings("unchecked")
public void testListOfStructsProjection() throws IOException {
Schema writeSchema =
@@ -587,7 +552,7 @@ public class TestRowProjection {
assertThat(projectedP2.isNullAt(0)).as("Should project null z").isTrue();
}
- @TestTemplate
+ @Test
public void testAddedFieldsWithRequiredChildren() throws Exception {
Schema schema = new Schema(Types.NestedField.required(1, "a", Types.LongType.get()));
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroReader.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroReader.java
deleted file mode 100644
index baae91dd18..0000000000
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroReader.java
+++ /dev/null
@@ -1,181 +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.data;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.function.Supplier;
-import org.apache.avro.LogicalType;
-import org.apache.avro.LogicalTypes;
-import org.apache.avro.Schema;
-import org.apache.avro.io.DatumReader;
-import org.apache.avro.io.Decoder;
-import org.apache.flink.table.data.RowData;
-import org.apache.iceberg.avro.AvroSchemaWithTypeVisitor;
-import org.apache.iceberg.avro.SupportsRowPosition;
-import org.apache.iceberg.avro.ValueReader;
-import org.apache.iceberg.avro.ValueReaders;
-import org.apache.iceberg.data.avro.DecoderResolver;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.iceberg.types.Type;
-import org.apache.iceberg.types.Types;
-
-/**
- * @deprecated will be removed in 1.8.0; use FlinkPlannedAvroReader instead.
- */
-@Deprecated
-public class FlinkAvroReader implements DatumReader<RowData>, SupportsRowPosition {
-
- private final Schema readSchema;
- private final ValueReader<RowData> reader;
- private Schema fileSchema = null;
-
- /**
- * @deprecated will be removed in 1.8.0; use FlinkPlannedAvroReader instead.
- */
- @Deprecated
- public FlinkAvroReader(org.apache.iceberg.Schema expectedSchema, Schema readSchema) {
- this(expectedSchema, readSchema, ImmutableMap.of());
- }
-
- /**
- * @deprecated will be removed in 1.8.0; use FlinkPlannedAvroReader instead.
- */
- @Deprecated
- @SuppressWarnings("unchecked")
- public FlinkAvroReader(
- org.apache.iceberg.Schema expectedSchema, Schema readSchema, Map<Integer, ?> constants) {
- this.readSchema = readSchema;
- this.reader =
- (ValueReader<RowData>)
- AvroSchemaWithTypeVisitor.visit(expectedSchema, readSchema, new ReadBuilder(constants));
- }
-
- @Override
- public void setSchema(Schema newFileSchema) {
- this.fileSchema = Schema.applyAliases(newFileSchema, readSchema);
- }
-
- @Override
- public RowData read(RowData reuse, Decoder decoder) throws IOException {
- return DecoderResolver.resolveAndRead(decoder, readSchema, fileSchema, reader, reuse);
- }
-
- @Override
- public void setRowPositionSupplier(Supplier<Long> posSupplier) {
- if (reader instanceof SupportsRowPosition) {
- ((SupportsRowPosition) reader).setRowPositionSupplier(posSupplier);
- }
- }
-
- private static class ReadBuilder extends AvroSchemaWithTypeVisitor<ValueReader<?>> {
- private final Map<Integer, ?> idToConstant;
-
- private ReadBuilder(Map<Integer, ?> idToConstant) {
- this.idToConstant = idToConstant;
- }
-
- @Override
- public ValueReader<?> record(
- Types.StructType expected, Schema record, List<String> names, List<ValueReader<?>> fields) {
- return FlinkValueReaders.struct(fields, expected.asStructType(), idToConstant);
- }
-
- @Override
- public ValueReader<?> union(Type expected, Schema union, List<ValueReader<?>> options) {
- return ValueReaders.union(options);
- }
-
- @Override
- public ValueReader<?> array(
- Types.ListType expected, Schema array, ValueReader<?> elementReader) {
- return FlinkValueReaders.array(elementReader);
- }
-
- @Override
- public ValueReader<?> map(
- Types.MapType expected, Schema map, ValueReader<?> keyReader, ValueReader<?> valueReader) {
- return FlinkValueReaders.arrayMap(keyReader, valueReader);
- }
-
- @Override
- public ValueReader<?> map(Types.MapType expected, Schema map, ValueReader<?> valueReader) {
- return FlinkValueReaders.map(FlinkValueReaders.strings(), valueReader);
- }
-
- @Override
- public ValueReader<?> primitive(Type.PrimitiveType expected, Schema primitive) {
- LogicalType logicalType = primitive.getLogicalType();
- if (logicalType != null) {
- switch (logicalType.getName()) {
- case "date":
- return ValueReaders.ints();
-
- case "time-micros":
- return FlinkValueReaders.timeMicros();
-
- case "timestamp-millis":
- return FlinkValueReaders.timestampMills();
-
- case "timestamp-micros":
- return FlinkValueReaders.timestampMicros();
-
- case "decimal":
- LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) logicalType;
- return FlinkValueReaders.decimal(
- ValueReaders.decimalBytesReader(primitive),
- decimal.getPrecision(),
- decimal.getScale());
-
- case "uuid":
- return FlinkValueReaders.uuids();
-
- default:
- throw new IllegalArgumentException("Unknown logical type: " + logicalType);
- }
- }
-
- switch (primitive.getType()) {
- case NULL:
- return ValueReaders.nulls();
- case BOOLEAN:
- return ValueReaders.booleans();
- case INT:
- return ValueReaders.ints();
- case LONG:
- return ValueReaders.longs();
- case FLOAT:
- return ValueReaders.floats();
- case DOUBLE:
- return ValueReaders.doubles();
- case STRING:
- return FlinkValueReaders.strings();
- case FIXED:
- return ValueReaders.fixed(primitive.getFixedSize());
- case BYTES:
- return ValueReaders.bytes();
- case ENUM:
- return FlinkValueReaders.enums(primitive.getEnumSymbols());
- default:
- throw new IllegalArgumentException("Unsupported type: " + primitive);
- }
- }
- }
-}
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroDeprecatedReaderWriter.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroDeprecatedReaderWriter.java
deleted file mode 100644
index 03910f4fda..0000000000
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkAvroDeprecatedReaderWriter.java
+++ /dev/null
@@ -1,38 +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.data;
-
-import java.io.File;
-import org.apache.iceberg.Files;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.avro.Avro;
-
-/**
- * @deprecated should be removed in 1.8.0; along with FlinkAvroReader.
- */
-@Deprecated
-public class TestFlinkAvroDeprecatedReaderWriter extends AbstractTestFlinkAvroReaderWriter {
-
- @Override
- protected Avro.ReadBuilder createAvroReadBuilder(File recordsFile, Schema schema) {
- return Avro.read(Files.localInput(recordsFile))
- .project(schema)
- .createReaderFunc(FlinkAvroReader::new);
- }
-}
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java
index f76e4c4942..3b98939ea1 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestRowProjection.java
@@ -24,9 +24,6 @@ import static org.assertj.core.api.Assertions.withPrecision;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.GenericMapData;
@@ -34,35 +31,21 @@ import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.iceberg.Files;
-import org.apache.iceberg.Parameter;
-import org.apache.iceberg.ParameterizedTestExtension;
-import org.apache.iceberg.Parameters;
import org.apache.iceberg.Schema;
import org.apache.iceberg.avro.Avro;
import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.types.Types;
-import org.junit.jupiter.api.TestTemplate;
-import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-@ExtendWith(ParameterizedTestExtension.class)
public class TestRowProjection {
@TempDir private Path temp;
- @Parameter(index = 0)
- protected Boolean useAvroPlannedReader;
-
- @Parameters(name = "useAvroPlannedReader={0}")
- protected static List<Object[]> parameters() {
- return Arrays.asList(new Object[] {Boolean.FALSE}, new Object[] {Boolean.TRUE});
- }
-
private RowData writeAndRead(String desc, Schema writeSchema, Schema readSchema, RowData row)
throws IOException {
File file = File.createTempFile("junit", desc + ".avro", temp.toFile());
@@ -79,20 +62,14 @@ public class TestRowProjection {
Avro.ReadBuilder builder =
Avro.read(Files.localInput(file))
.project(readSchema)
- .createReaderFunc(FlinkAvroReader::new);
- if (useAvroPlannedReader) {
- builder =
- Avro.read(Files.localInput(file))
- .project(readSchema)
- .createResolvingReader(FlinkPlannedAvroReader::create);
- }
+ .createResolvingReader(FlinkPlannedAvroReader::create);
Iterable<RowData> records = builder.build();
return Iterables.getOnlyElement(records);
}
- @TestTemplate
+ @Test
public void testFullProjection() throws Exception {
Schema schema =
new Schema(
@@ -107,7 +84,7 @@ public class TestRowProjection {
assertThat(projected.getString(1)).asString().isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testSpecialCharacterProjection() throws Exception {
Schema schema =
new Schema(
@@ -127,7 +104,7 @@ public class TestRowProjection {
assertThat(projected.getString(0)).asString().isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testReorderedFullProjection() throws Exception {
Schema schema =
new Schema(
@@ -147,7 +124,7 @@ public class TestRowProjection {
assertThat(projected.getLong(1)).isEqualTo(34);
}
- @TestTemplate
+ @Test
public void testReorderedProjection() throws Exception {
Schema schema =
new Schema(
@@ -169,7 +146,7 @@ public class TestRowProjection {
assertThat(projected.isNullAt(2)).isTrue();
}
- @TestTemplate
+ @Test
public void testRenamedAddedField() throws Exception {
Schema schema =
new Schema(
@@ -199,7 +176,7 @@ public class TestRowProjection {
assertThat(projected.isNullAt(3)).as("Should contain empty value on new column 4").isTrue();
}
- @TestTemplate
+ @Test
public void testEmptyProjection() throws Exception {
Schema schema =
new Schema(
@@ -214,7 +191,7 @@ public class TestRowProjection {
assertThat(projected.getArity()).isEqualTo(0);
}
- @TestTemplate
+ @Test
public void testBasicProjection() throws Exception {
Schema writeSchema =
new Schema(
@@ -238,7 +215,7 @@ public class TestRowProjection {
assertThat(projected.getString(0)).asString().isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testRename() throws Exception {
Schema writeSchema =
new Schema(
@@ -261,7 +238,7 @@ public class TestRowProjection {
.isEqualTo("test");
}
- @TestTemplate
+ @Test
public void testNestedStructProjection() throws Exception {
Schema writeSchema =
new Schema(
@@ -327,7 +304,7 @@ public class TestRowProjection {
.isEqualTo(-1.539054f, withPrecision(0.000001f));
}
- @TestTemplate
+ @Test
public void testMapProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -369,19 +346,7 @@ public class TestRowProjection {
assertThat(projected.getMap(0)).isEqualTo(properties);
}
- private Map<String, ?> toStringMap(Map<?, ?> map) {
- Map<String, Object> stringMap = Maps.newHashMap();
- for (Map.Entry<?, ?> entry : map.entrySet()) {
- if (entry.getValue() instanceof CharSequence) {
- stringMap.put(entry.getKey().toString(), entry.getValue().toString());
- } else {
- stringMap.put(entry.getKey().toString(), entry.getValue());
- }
- }
- return stringMap;
- }
-
- @TestTemplate
+ @Test
public void testMapOfStructsProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -481,7 +446,7 @@ public class TestRowProjection {
.isEqualTo(52.995143f, withPrecision(0.000001f));
}
- @TestTemplate
+ @Test
public void testListProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -510,8 +475,7 @@ public class TestRowProjection {
assertThat(projected.getArray(0)).isEqualTo(values);
}
- @TestTemplate
- @SuppressWarnings("unchecked")
+ @Test
public void testListOfStructsProjection() throws IOException {
Schema writeSchema =
new Schema(
@@ -587,7 +551,7 @@ public class TestRowProjection {
assertThat(projectedP2.isNullAt(0)).as("Should project null z").isTrue();
}
- @TestTemplate
+ @Test
public void testAddedFieldsWithRequiredChildren() throws Exception {
Schema schema = new Schema(Types.NestedField.required(1, "a", Types.LongType.get()));
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/data/SparkAvroReader.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/data/SparkAvroReader.java
deleted file mode 100644
index 7d92d963a9..0000000000
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/data/SparkAvroReader.java
+++ /dev/null
@@ -1,180 +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.spark.data;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.function.Supplier;
-import org.apache.avro.LogicalType;
-import org.apache.avro.LogicalTypes;
-import org.apache.avro.Schema;
-import org.apache.avro.io.DatumReader;
-import org.apache.avro.io.Decoder;
-import org.apache.iceberg.avro.AvroSchemaWithTypeVisitor;
-import org.apache.iceberg.avro.SupportsRowPosition;
-import org.apache.iceberg.avro.ValueReader;
-import org.apache.iceberg.avro.ValueReaders;
-import org.apache.iceberg.data.avro.DecoderResolver;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.iceberg.types.Type;
-import org.apache.iceberg.types.Types;
-import org.apache.spark.sql.catalyst.InternalRow;
-
-/**
- * @deprecated will be removed in 1.8.0; use SparkPlannedAvroReader instead.
- */
-@Deprecated
-public class SparkAvroReader implements DatumReader<InternalRow>, SupportsRowPosition {
-
- private final Schema readSchema;
- private final ValueReader<InternalRow> reader;
- private Schema fileSchema = null;
-
- /**
- * @deprecated will be removed in 1.8.0; use SparkPlannedAvroReader instead.
- */
- @Deprecated
- public SparkAvroReader(org.apache.iceberg.Schema expectedSchema, Schema readSchema) {
- this(expectedSchema, readSchema, ImmutableMap.of());
- }
-
- /**
- * @deprecated will be removed in 1.8.0; use SparkPlannedAvroReader instead.
- */
- @Deprecated
- @SuppressWarnings("unchecked")
- public SparkAvroReader(
- org.apache.iceberg.Schema expectedSchema, Schema readSchema, Map<Integer, ?> constants) {
- this.readSchema = readSchema;
- this.reader =
- (ValueReader<InternalRow>)
- AvroSchemaWithTypeVisitor.visit(expectedSchema, readSchema, new ReadBuilder(constants));
- }
-
- @Override
- public void setSchema(Schema newFileSchema) {
- this.fileSchema = Schema.applyAliases(newFileSchema, readSchema);
- }
-
- @Override
- public InternalRow read(InternalRow reuse, Decoder decoder) throws IOException {
- return DecoderResolver.resolveAndRead(decoder, readSchema, fileSchema, reader, reuse);
- }
-
- @Override
- public void setRowPositionSupplier(Supplier<Long> posSupplier) {
- if (reader instanceof SupportsRowPosition) {
- ((SupportsRowPosition) reader).setRowPositionSupplier(posSupplier);
- }
- }
-
- private static class ReadBuilder extends AvroSchemaWithTypeVisitor<ValueReader<?>> {
- private final Map<Integer, ?> idToConstant;
-
- private ReadBuilder(Map<Integer, ?> idToConstant) {
- this.idToConstant = idToConstant;
- }
-
- @Override
- public ValueReader<?> record(
- Types.StructType expected, Schema record, List<String> names, List<ValueReader<?>> fields) {
- return SparkValueReaders.struct(fields, expected, idToConstant);
- }
-
- @Override
- public ValueReader<?> union(Type expected, Schema union, List<ValueReader<?>> options) {
- return ValueReaders.union(options);
- }
-
- @Override
- public ValueReader<?> array(
- Types.ListType expected, Schema array, ValueReader<?> elementReader) {
- return SparkValueReaders.array(elementReader);
- }
-
- @Override
- public ValueReader<?> map(
- Types.MapType expected, Schema map, ValueReader<?> keyReader, ValueReader<?> valueReader) {
- return SparkValueReaders.arrayMap(keyReader, valueReader);
- }
-
- @Override
- public ValueReader<?> map(Types.MapType expected, Schema map, ValueReader<?> valueReader) {
- return SparkValueReaders.map(SparkValueReaders.strings(), valueReader);
- }
-
- @Override
- public ValueReader<?> primitive(Type.PrimitiveType expected, Schema primitive) {
- LogicalType logicalType = primitive.getLogicalType();
- if (logicalType != null) {
- switch (logicalType.getName()) {
- case "date":
- // Spark uses the same representation
- return ValueReaders.ints();
-
- case "timestamp-millis":
- // adjust to microseconds
- ValueReader<Long> longs = ValueReaders.longs();
- return (ValueReader<Long>) (decoder, ignored) -> longs.read(decoder, null) * 1000L;
-
- case "timestamp-micros":
- // Spark uses the same representation
- return ValueReaders.longs();
-
- case "decimal":
- return SparkValueReaders.decimal(
- ValueReaders.decimalBytesReader(primitive),
- ((LogicalTypes.Decimal) logicalType).getScale());
-
- case "uuid":
- return SparkValueReaders.uuids();
-
- default:
- throw new IllegalArgumentException("Unknown logical type: " + logicalType);
- }
- }
-
- switch (primitive.getType()) {
- case NULL:
- return ValueReaders.nulls();
- case BOOLEAN:
- return ValueReaders.booleans();
- case INT:
- return ValueReaders.ints();
- case LONG:
- return ValueReaders.longs();
- case FLOAT:
- return ValueReaders.floats();
- case DOUBLE:
- return ValueReaders.doubles();
- case STRING:
- return SparkValueReaders.strings();
- case FIXED:
- return ValueReaders.fixed(primitive.getFixedSize());
- case BYTES:
- return ValueReaders.bytes();
- case ENUM:
- return SparkValueReaders.enums(primitive.getEnumSymbols());
- default:
- throw new IllegalArgumentException("Unsupported type: " + primitive);
- }
- }
- }
-}