You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fc...@apache.org on 2024/10/11 09:52:01 UTC
(flink) branch release-1.20 updated: [FLINK-36468][parquet] Use Flink Preconditions util instead of Parquet
This is an automated email from the ASF dual-hosted git repository.
fcsaky pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new 74eeba628fa [FLINK-36468][parquet] Use Flink Preconditions util instead of Parquet
74eeba628fa is described below
commit 74eeba628fa32c7e23fe4edb51233c370202a9fd
Author: Ferenc Csaky <fe...@pm.me>
AuthorDate: Fri Oct 11 11:51:54 2024 +0200
[FLINK-36468][parquet] Use Flink Preconditions util instead of Parquet
---
.../apache/flink/formats/parquet/PositionOutputStreamAdapter.java | 2 +-
.../formats/parquet/utils/ParquetFormatStatisticsReportUtil.java | 5 +++--
.../apache/flink/formats/parquet/vector/ParquetDecimalVector.java | 4 ++--
.../flink/formats/parquet/vector/ParquetSplitReaderUtil.java | 2 +-
.../flink/formats/parquet/vector/reader/AbstractColumnReader.java | 4 ++--
.../flink/formats/parquet/vector/reader/RunLengthDecoder.java | 6 +++---
.../formats/parquet/vector/reader/TimestampColumnReader.java | 8 ++++----
7 files changed, 16 insertions(+), 15 deletions(-)
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/PositionOutputStreamAdapter.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/PositionOutputStreamAdapter.java
index 4fc733166bd..f5336293140 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/PositionOutputStreamAdapter.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/PositionOutputStreamAdapter.java
@@ -25,7 +25,7 @@ import org.apache.parquet.io.PositionOutputStream;
import java.io.IOException;
-import static org.apache.parquet.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
/** An adapter to turn Flink's {@link FSDataOutputStream} into a {@link PositionOutputStream}. */
@Internal
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetFormatStatisticsReportUtil.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetFormatStatisticsReportUtil.java
index a40784db312..9eed138bab7 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetFormatStatisticsReportUtil.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetFormatStatisticsReportUtil.java
@@ -31,7 +31,6 @@ import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.parquet.Preconditions;
import org.apache.parquet.column.statistics.BinaryStatistics;
import org.apache.parquet.column.statistics.DoubleStatistics;
import org.apache.parquet.column.statistics.FloatStatistics;
@@ -64,6 +63,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
/** Utils for Parquet format statistics report. */
public class ParquetFormatStatisticsReportUtil {
@@ -316,7 +317,7 @@ public class ParquetFormatStatisticsReportUtil {
}
private static Timestamp binaryToTimestamp(Binary timestamp, boolean utcTimestamp) {
- Preconditions.checkArgument(timestamp.length() == 12, "Must be 12 bytes");
+ checkArgument(timestamp.length() == 12, "Must be 12 bytes");
ByteBuffer buf = timestamp.toByteBuffer();
buf.order(ByteOrder.LITTLE_ENDIAN);
long timeOfDayNanos = buf.getLong();
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDecimalVector.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDecimalVector.java
index 0c62dbb0460..62395d3c1f2 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDecimalVector.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDecimalVector.java
@@ -27,7 +27,7 @@ import org.apache.flink.table.data.columnar.vector.DecimalColumnVector;
import org.apache.flink.table.data.columnar.vector.IntColumnVector;
import org.apache.flink.table.data.columnar.vector.LongColumnVector;
-import org.apache.parquet.Preconditions;
+import static org.apache.flink.util.Preconditions.checkArgument;
/**
* Parquet write decimal as int32 and int64 and binary, this class wrap the real vector to provide
@@ -52,7 +52,7 @@ public class ParquetDecimalVector implements DecimalColumnVector {
return DecimalData.fromUnscaledLong(
((LongColumnVector) vector).getLong(i), precision, scale);
} else {
- Preconditions.checkArgument(
+ checkArgument(
vector instanceof BytesColumnVector,
"Reading decimal type occur unsupported vector type: %s",
vector.getClass());
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
index 77fa87c7e26..77eee0e03a4 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
@@ -86,7 +86,7 @@ import java.util.Map;
import java.util.stream.Collectors;
import static org.apache.flink.table.utils.DateTimeUtils.toInternal;
-import static org.apache.parquet.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkArgument;
/** Util for generating {@link ParquetColumnarRowSplitReader}. */
public class ParquetSplitReaderUtil {
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/AbstractColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/AbstractColumnReader.java
index c57a5afc6d1..aefcc8fe93d 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/AbstractColumnReader.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/AbstractColumnReader.java
@@ -21,7 +21,6 @@ import org.apache.flink.formats.parquet.vector.ParquetDictionary;
import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
import org.apache.flink.table.data.columnar.vector.writable.WritableIntVector;
-import org.apache.parquet.Preconditions;
import org.apache.parquet.bytes.ByteBufferInputStream;
import org.apache.parquet.bytes.BytesInput;
import org.apache.parquet.bytes.BytesUtils;
@@ -44,6 +43,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
+import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
/**
@@ -132,7 +132,7 @@ public abstract class AbstractColumnReader<VECTOR extends WritableColumnVector>
protected void checkTypeName(PrimitiveType.PrimitiveTypeName expectedName) {
PrimitiveType.PrimitiveTypeName actualName =
descriptor.getPrimitiveType().getPrimitiveTypeName();
- Preconditions.checkArgument(
+ checkArgument(
actualName == expectedName,
"Expected type name: %s, actual type name: %s",
expectedName,
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RunLengthDecoder.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RunLengthDecoder.java
index 1492e11409b..275bad1c222 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RunLengthDecoder.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RunLengthDecoder.java
@@ -20,7 +20,6 @@ package org.apache.flink.formats.parquet.vector.reader;
import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
import org.apache.flink.table.data.columnar.vector.writable.WritableIntVector;
-import org.apache.parquet.Preconditions;
import org.apache.parquet.bytes.ByteBufferInputStream;
import org.apache.parquet.bytes.BytesUtils;
import org.apache.parquet.column.values.bitpacking.BytePacker;
@@ -31,6 +30,8 @@ import org.apache.parquet.io.ParquetDecodingException;
import java.io.IOException;
import java.nio.ByteBuffer;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
/**
* Run length decoder for data and dictionary ids. See
* https://github.com/apache/parquet-format/blob/master/Encodings.md See {@link
@@ -107,8 +108,7 @@ final class RunLengthDecoder {
/** Initializes the internal state for decoding ints of `bitWidth`. */
private void initWidthAndPacker(int bitWidth) {
- Preconditions.checkArgument(
- bitWidth >= 0 && bitWidth <= 32, "bitWidth must be >= 0 and <= 32");
+ checkArgument(bitWidth >= 0 && bitWidth <= 32, "bitWidth must be >= 0 and <= 32");
this.bitWidth = bitWidth;
this.bytesWidth = BytesUtils.paddedByteCountFromBits(bitWidth);
this.packer = Packer.LITTLE_ENDIAN.newBytePacker(bitWidth);
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/TimestampColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/TimestampColumnReader.java
index 7a36edc573b..5abfcea3d55 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/TimestampColumnReader.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/TimestampColumnReader.java
@@ -21,7 +21,6 @@ import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.columnar.vector.writable.WritableIntVector;
import org.apache.flink.table.data.columnar.vector.writable.WritableTimestampVector;
-import org.apache.parquet.Preconditions;
import org.apache.parquet.column.ColumnDescriptor;
import org.apache.parquet.column.page.PageReader;
import org.apache.parquet.io.api.Binary;
@@ -34,6 +33,8 @@ import java.nio.ByteOrder;
import java.sql.Timestamp;
import java.util.concurrent.TimeUnit;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
/**
* Timestamp {@link ColumnReader}. We support INT96 and INT64 now, julianDay(4) + nanosOfDay(8). See
* https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#timestamp TIMESTAMP_MILLIS
@@ -72,7 +73,7 @@ public class TimestampColumnReader extends AbstractColumnReader<WritableTimestam
}
private void checkTypeName() {
- Preconditions.checkArgument(
+ checkArgument(
actualName == PrimitiveType.PrimitiveTypeName.INT96
|| actualName == PrimitiveType.PrimitiveTypeName.INT64,
"Expected type name: %s or %s, actual type name: %s",
@@ -137,8 +138,7 @@ public class TimestampColumnReader extends AbstractColumnReader<WritableTimestam
public static TimestampData decodeInt96ToTimestamp(
boolean utcTimestamp, org.apache.parquet.column.Dictionary dictionary, int id) {
Binary binary = dictionary.decodeToBinary(id);
- Preconditions.checkArgument(
- binary.length() == 12, "Timestamp with int96 should be 12 bytes.");
+ checkArgument(binary.length() == 12, "Timestamp with int96 should be 12 bytes.");
ByteBuffer buffer = binary.toByteBuffer().order(ByteOrder.LITTLE_ENDIAN);
return int96ToTimestamp(utcTimestamp, buffer.getLong(), buffer.getInt());
}