You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2023/02/14 11:44:19 UTC
[flink-table-store] branch master updated: [FLINK-31054] Flink free to common codegen core shade
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 6a5e4084 [FLINK-31054] Flink free to common codegen core shade
6a5e4084 is described below
commit 6a5e4084d95563f7094deec62c36aa1e921f9ec6
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Tue Feb 14 19:44:14 2023 +0800
[FLINK-31054] Flink free to common codegen core shade
This closes #532
---
flink-table-store-codegen-loader/pom.xml | 7 -
flink-table-store-codegen/pom.xml | 6 -
flink-table-store-common/pom.xml | 15 +-
.../flink/table/store/codegen/CompileUtils.java | 4 +-
.../codegen/codesplit/BlockStatementGrouper.java | 2 +-
.../flink/table/store/data/GenericArray.java | 3 +-
.../table/store/data/RandomAccessInputView.java | 4 +-
.../data/columnar/heap/AbstractHeapVector.java | 4 +-
.../store/data/serializer/ListSerializer.java | 2 +-
.../store/data/serializer/VersionedSerializer.java | 53 ++
.../table/store/file/predicate/CompareUtils.java | 18 +-
.../store/file/utils/RecordReaderIterator.java | 2 +-
.../java/org/apache/flink/table/store/fs/Path.java | 2 +-
.../flink/table/store/options/ConfigOption.java | 2 +-
.../store/options/description/TextElement.java | 4 +-
.../table/store/plugin/ComponentClassLoader.java | 2 +-
.../apache/flink/table/store/utils/ArrayUtils.java | 357 +++++++++++++
.../flink/table/store/utils/CloseableIterator.java | 191 +++++++
.../flink/table/store/utils/ExceptionUtils.java | 576 +++++++++++++++++++++
.../FunctionWithException.java} | 37 +-
.../apache/flink/table/store/utils/IOUtils.java | 9 +-
.../flink/table/store/utils/LongCounter.java | 63 +++
.../org/apache/flink/table/store/utils/Pair.java | 98 ++++
.../SerializableFunction.java} | 26 +-
.../flink/table/store/utils/StringUtils.java | 295 ++++++++++-
.../apache/flink/table/store/utils/TimeUtils.java | 12 -
.../codegen/codesplit/CodeRewriterTestBase.java | 6 +-
.../store/codegen/codesplit/CodeSplitTestUtil.java | 10 +-
.../codegen/codesplit/JavaCodeSplitterTest.java | 6 +-
.../flink/table/store/data/DataFormatTestUtil.java | 11 +-
.../store/data/serializer/SerializerTestBase.java | 2 +-
.../table/store/datagen/DataGeneratorMapper.java | 2 +-
.../connector/VersionedSerializerWrapper.java | 30 +-
.../table/store/connector/sink/FileStoreSink.java | 4 +-
.../connector/sink/CommitterOperatorTest.java | 5 +-
.../connector/sink/CommitterOperatorTestBase.java | 2 +-
flink-table-store-core/pom.xml | 7 -
.../table/store/file/append/AppendOnlyWriter.java | 4 +-
.../table/store/file/catalog/AbstractCatalog.java | 3 +-
.../flink/table/store/file/io/FileWriter.java | 2 +-
.../table/store/file/io/RowDataFileWriter.java | 6 +-
.../store/file/io/RowDataRollingFileWriter.java | 2 +-
.../manifest/ManifestCommittableSerializer.java | 7 +-
.../table/store/file/manifest/ManifestList.java | 6 +-
.../file/operation/AppendOnlyFileStoreWrite.java | 2 +-
.../store/file/sort/BinaryExternalSortBuffer.java | 2 +-
.../file/utils/FatalExitExceptionHandler.java | 5 +-
.../store/file/utils/FileStorePathFactory.java | 2 +-
.../table/store/file/utils/PartitionPathUtils.java | 2 +-
.../table/store/table/AbstractFileStoreTable.java | 2 +-
.../table/sink/FileCommittableSerializer.java | 6 +-
.../store/file/io/KeyValueFileReadWriteTest.java | 2 +-
.../table/store/file/io/RollingFileWriterTest.java | 2 +-
flink-table-store-format/pom.xml | 7 -
.../table/store/format/avro/AvroBulkWriter.java | 4 +-
.../table/store/format/orc/OrcFileFormat.java | 7 +-
.../table/store/format/orc/OrcReaderFactory.java | 12 +-
.../table/store/format/orc/OrcWriterFactory.java | 3 +-
.../orc/SerializableHadoopConfigWrapper.java | 4 +-
.../store/format/orc/writer/OrcBulkWriter.java | 3 +-
.../store/format/parquet/ParquetFileFormat.java | 5 +-
.../format/parquet/writer/ParquetBulkWriter.java | 3 +-
.../store/format/parquet/ParquetReadWriteTest.java | 2 -
.../flink-table-store-hive-connector/pom.xml | 14 +
.../flink/table/store/TableStoreJobConf.java | 12 +-
.../apache/flink/table/store/hive/HiveSchema.java | 2 +-
flink-table-store-shade/pom.xml | 127 +----
pom.xml | 1 +
68 files changed, 1790 insertions(+), 350 deletions(-)
diff --git a/flink-table-store-codegen-loader/pom.xml b/flink-table-store-codegen-loader/pom.xml
index dabce6d0..44b28168 100644
--- a/flink-table-store-codegen-loader/pom.xml
+++ b/flink-table-store-codegen-loader/pom.xml
@@ -32,13 +32,6 @@ under the License.
<name>Flink Table Store : Code Gen Loader</name>
<dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-store-common</artifactId>
diff --git a/flink-table-store-codegen/pom.xml b/flink-table-store-codegen/pom.xml
index aaf0539d..5940493c 100644
--- a/flink-table-store-codegen/pom.xml
+++ b/flink-table-store-codegen/pom.xml
@@ -38,12 +38,6 @@ under the License.
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${flink.version}</version>
- </dependency>
-
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
diff --git a/flink-table-store-common/pom.xml b/flink-table-store-common/pom.xml
index f67a1be2..45c3ed4d 100644
--- a/flink-table-store-common/pom.xml
+++ b/flink-table-store-common/pom.xml
@@ -38,15 +38,14 @@ under the License.
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-table-code-splitter</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
+ <artifactId>flink-shaded-jackson</artifactId>
+ <version>${flink.shaded.jackson.version}-${flink.shaded.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-jackson</artifactId>
- <version>${flink.shaded.jackson.version}-${flink.shaded.version}</version>
+ <artifactId>flink-shaded-guava</artifactId>
+ <version>${flink.shaded.guava.version}-${flink.shaded.version}</version>
</dependency>
<!-- Java compiler -->
@@ -197,12 +196,6 @@ under the License.
</exclusion>
</exclusions>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-common</artifactId>
- <version>1.16.0</version>
- <scope>test</scope>
- </dependency>
</dependencies>
<build>
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/codegen/CompileUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/codegen/CompileUtils.java
index 602fabe0..1a738f3c 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/codegen/CompileUtils.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/codegen/CompileUtils.java
@@ -18,8 +18,6 @@
package org.apache.flink.table.store.codegen;
-import org.apache.flink.api.common.InvalidProgramException;
-
import org.apache.flink.shaded.guava30.com.google.common.cache.Cache;
import org.apache.flink.shaded.guava30.com.google.common.cache.CacheBuilder;
@@ -85,7 +83,7 @@ public final class CompileUtils {
compiler.cook(code);
} catch (Throwable t) {
System.out.println(addLineNumber(code));
- throw new InvalidProgramException(
+ throw new RuntimeException(
"Table program cannot be compiled. This is a bug. Please file an issue.", t);
}
try {
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/codegen/codesplit/BlockStatementGrouper.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/codegen/codesplit/BlockStatementGrouper.java
index 1af59e00..941b0ef4 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/codegen/codesplit/BlockStatementGrouper.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/codegen/codesplit/BlockStatementGrouper.java
@@ -19,6 +19,7 @@ package org.apache.flink.table.store.codegen.codesplit;
import org.apache.flink.table.store.codegen.codesplit.JavaParser.BlockStatementContext;
import org.apache.flink.table.store.codegen.codesplit.JavaParser.StatementContext;
+import org.apache.flink.table.store.utils.Pair;
import org.antlr.v4.runtime.CharStreams;
import org.antlr.v4.runtime.CommonTokenStream;
@@ -26,7 +27,6 @@ import org.antlr.v4.runtime.ParserRuleContext;
import org.antlr.v4.runtime.Token;
import org.antlr.v4.runtime.TokenStreamRewriter;
import org.antlr.v4.runtime.atn.PredictionMode;
-import org.apache.commons.lang3.tuple.Pair;
import java.util.ArrayList;
import java.util.HashMap;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericArray.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericArray.java
index ecb42bd0..19fc2851 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericArray.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericArray.java
@@ -19,8 +19,7 @@
package org.apache.flink.table.store.data;
import org.apache.flink.table.store.types.ArrayType;
-
-import org.apache.commons.lang3.ArrayUtils;
+import org.apache.flink.table.store.utils.ArrayUtils;
import java.util.Arrays;
import java.util.Objects;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/RandomAccessInputView.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/RandomAccessInputView.java
index 3b963df3..da69a9a8 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/RandomAccessInputView.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/RandomAccessInputView.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.data;
-import org.apache.flink.core.memory.SeekableDataInputView;
import org.apache.flink.table.store.memory.MemorySegment;
import org.apache.flink.table.store.utils.MathUtils;
@@ -26,7 +25,7 @@ import java.io.EOFException;
import java.util.ArrayList;
/** A {@link AbstractPagedInputView} to read pages in memory. */
-public class RandomAccessInputView extends AbstractPagedInputView implements SeekableDataInputView {
+public class RandomAccessInputView extends AbstractPagedInputView {
private final ArrayList<MemorySegment> segments;
@@ -55,7 +54,6 @@ public class RandomAccessInputView extends AbstractPagedInputView implements See
this.limitInLastSegment = limitInLastSegment;
}
- @Override
public void setReadPosition(long position) {
final int bufferNum = (int) (position >>> this.segmentSizeBits);
final int offset = (int) (position & this.segmentSizeMask);
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/heap/AbstractHeapVector.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/heap/AbstractHeapVector.java
index 7ab66daf..72c9adaa 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/heap/AbstractHeapVector.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/heap/AbstractHeapVector.java
@@ -18,8 +18,8 @@
package org.apache.flink.table.store.data.columnar.heap;
-import org.apache.flink.core.memory.MemoryUtils;
import org.apache.flink.table.store.data.columnar.writable.AbstractWritableVector;
+import org.apache.flink.table.store.memory.MemorySegment;
import java.nio.ByteOrder;
import java.util.Arrays;
@@ -29,7 +29,7 @@ public abstract class AbstractHeapVector extends AbstractWritableVector {
public static final boolean LITTLE_ENDIAN = ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN;
- public static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
+ public static final sun.misc.Unsafe UNSAFE = MemorySegment.UNSAFE;
public static final int BYTE_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
public static final int INT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(int[].class);
public static final int LONG_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(long[].class);
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/serializer/ListSerializer.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/serializer/ListSerializer.java
index c5075d67..f6e3906c 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/serializer/ListSerializer.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/serializer/ListSerializer.java
@@ -25,7 +25,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
/**
* A serializer for {@link List Lists}. The serializer relies on an element serializer for the
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/serializer/VersionedSerializer.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/serializer/VersionedSerializer.java
new file mode 100644
index 00000000..9faedb2d
--- /dev/null
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/serializer/VersionedSerializer.java
@@ -0,0 +1,53 @@
+/*
+ * 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.flink.table.store.data.serializer;
+
+import java.io.IOException;
+
+/** A simple serializer interface for versioned serialization. */
+public interface VersionedSerializer<T> {
+
+ /**
+ * Gets the version with which this serializer serializes.
+ *
+ * @return The version of the serialization schema.
+ */
+ int getVersion();
+
+ /**
+ * Serializes the given object. The serialization is assumed to correspond to the current
+ * serialization version (as returned by {@link #getVersion()}.
+ *
+ * @param obj The object to serialize.
+ * @return The serialized data (bytes).
+ * @throws IOException Thrown, if the serialization fails.
+ */
+ byte[] serialize(T obj) throws IOException;
+
+ /**
+ * De-serializes the given data (bytes) which was serialized with the scheme of the indicated
+ * version.
+ *
+ * @param version The version in which the data was serialized
+ * @param serialized The serialized data
+ * @return The deserialized object
+ * @throws IOException Thrown, if the deserialization fails.
+ */
+ T deserialize(int version, byte[] serialized) throws IOException;
+}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java
index 7a153d54..10d28290 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java
@@ -18,23 +18,31 @@
package org.apache.flink.table.store.file.predicate;
-import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArrayComparator;
import org.apache.flink.table.store.types.DataType;
+import static java.lang.Math.min;
+
/** Utils for comparator. */
public class CompareUtils {
private CompareUtils() {}
- private static final BytePrimitiveArrayComparator BINARY_COMPARATOR =
- new BytePrimitiveArrayComparator(true);
-
public static int compareLiteral(DataType type, Object v1, Object v2) {
if (v1 instanceof Comparable) {
return ((Comparable<Object>) v1).compareTo(v2);
} else if (v1 instanceof byte[]) {
- return BINARY_COMPARATOR.compare((byte[]) v1, (byte[]) v2);
+ return compare((byte[]) v1, (byte[]) v2);
} else {
throw new RuntimeException("Unsupported type: " + type);
}
}
+
+ private static int compare(byte[] first, byte[] second) {
+ for (int x = 0; x < min(first.length, second.length); x++) {
+ int cmp = first[x] - second[x];
+ if (cmp != 0) {
+ return cmp;
+ }
+ }
+ return first.length - second.length;
+ }
}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/utils/RecordReaderIterator.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/utils/RecordReaderIterator.java
index b3420070..29e9bf7c 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/utils/RecordReaderIterator.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/utils/RecordReaderIterator.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.file.utils;
-import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.table.store.utils.CloseableIterator;
import java.io.IOException;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/fs/Path.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/fs/Path.java
index e5970266..363baa0c 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/fs/Path.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/fs/Path.java
@@ -22,7 +22,7 @@
package org.apache.flink.table.store.fs;
-import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.table.store.utils.StringUtils;
import java.io.Serializable;
import java.net.URI;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/options/ConfigOption.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/options/ConfigOption.java
index 7c97355f..5cbce081 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/options/ConfigOption.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/options/ConfigOption.java
@@ -25,7 +25,7 @@ import java.util.Collections;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
/**
* A {@code ConfigOption} describes a configuration parameter. It encapsulates the configuration
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/options/description/TextElement.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/options/description/TextElement.java
index 0264cb54..7ed595d7 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/options/description/TextElement.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/options/description/TextElement.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.options.description;
-import org.apache.flink.shaded.guava30.com.google.common.base.Strings;
+import org.apache.flink.table.store.utils.StringUtils;
import java.util.Arrays;
import java.util.Collections;
@@ -57,7 +57,7 @@ public class TextElement implements BlockElement, InlineElement {
/** Wraps a list of {@link InlineElement}s into a single {@link TextElement}. */
public static InlineElement wrap(InlineElement... elements) {
- return text(Strings.repeat("%s", elements.length), elements);
+ return text(StringUtils.repeat("%s", elements.length), elements);
}
/**
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/plugin/ComponentClassLoader.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/plugin/ComponentClassLoader.java
index 14c4c657..46598f9c 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/plugin/ComponentClassLoader.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/plugin/ComponentClassLoader.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.plugin;
import org.apache.flink.table.store.annotation.VisibleForTesting;
-import org.apache.flink.util.function.FunctionWithException;
+import org.apache.flink.table.store.utils.FunctionWithException;
import org.apache.flink.shaded.guava30.com.google.common.collect.Iterators;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/ArrayUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/ArrayUtils.java
new file mode 100644
index 00000000..fe24037f
--- /dev/null
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/ArrayUtils.java
@@ -0,0 +1,357 @@
+/*
+ * 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.flink.table.store.utils;
+
+/** Utils for array. */
+public class ArrayUtils {
+
+ /** An empty immutable {@code String} array. */
+ public static final String[] EMPTY_STRING_ARRAY = new String[0];
+
+ /** An empty immutable {@code long} array. */
+ public static final long[] EMPTY_LONG_ARRAY = new long[0];
+ /** An empty immutable {@code Long} array. */
+ public static final Long[] EMPTY_LONG_OBJECT_ARRAY = new Long[0];
+ /** An empty immutable {@code int} array. */
+ public static final int[] EMPTY_INT_ARRAY = new int[0];
+ /** An empty immutable {@code Integer} array. */
+ public static final Integer[] EMPTY_INTEGER_OBJECT_ARRAY = new Integer[0];
+ /** An empty immutable {@code short} array. */
+ public static final short[] EMPTY_SHORT_ARRAY = new short[0];
+ /** An empty immutable {@code Short} array. */
+ public static final Short[] EMPTY_SHORT_OBJECT_ARRAY = new Short[0];
+ /** An empty immutable {@code byte} array. */
+ public static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
+ /** An empty immutable {@code Byte} array. */
+ public static final Byte[] EMPTY_BYTE_OBJECT_ARRAY = new Byte[0];
+ /** An empty immutable {@code double} array. */
+ public static final double[] EMPTY_DOUBLE_ARRAY = new double[0];
+ /** An empty immutable {@code Double} array. */
+ public static final Double[] EMPTY_DOUBLE_OBJECT_ARRAY = new Double[0];
+ /** An empty immutable {@code float} array. */
+ public static final float[] EMPTY_FLOAT_ARRAY = new float[0];
+ /** An empty immutable {@code Float} array. */
+ public static final Float[] EMPTY_FLOAT_OBJECT_ARRAY = new Float[0];
+ /** An empty immutable {@code boolean} array. */
+ public static final boolean[] EMPTY_BOOLEAN_ARRAY = new boolean[0];
+ /** An empty immutable {@code Boolean} array. */
+ public static final Boolean[] EMPTY_BOOLEAN_OBJECT_ARRAY = new Boolean[0];
+
+ private ArrayUtils() {}
+
+ /**
+ * Converts an array of primitive booleans to objects.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code boolean} array
+ * @return a {@code Boolean} array, {@code null} if null array input
+ */
+ public static Boolean[] toObject(final boolean[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_BOOLEAN_OBJECT_ARRAY;
+ }
+ final Boolean[] result = new Boolean[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = (array[i] ? Boolean.TRUE : Boolean.FALSE);
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of primitive bytes to objects.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code byte} array
+ * @return a {@code Byte} array, {@code null} if null array input
+ */
+ public static Byte[] toObject(final byte[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_BYTE_OBJECT_ARRAY;
+ }
+ final Byte[] result = new Byte[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of primitive doubles to objects.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code double} array
+ * @return a {@code Double} array, {@code null} if null array input
+ */
+ public static Double[] toObject(final double[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_DOUBLE_OBJECT_ARRAY;
+ }
+ final Double[] result = new Double[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of primitive floats to objects.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code float} array
+ * @return a {@code Float} array, {@code null} if null array input
+ */
+ public static Float[] toObject(final float[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_FLOAT_OBJECT_ARRAY;
+ }
+ final Float[] result = new Float[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of primitive ints to objects.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array an {@code int} array
+ * @return an {@code Integer} array, {@code null} if null array input
+ */
+ public static Integer[] toObject(final int[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_INTEGER_OBJECT_ARRAY;
+ }
+ final Integer[] result = new Integer[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of primitive longs to objects.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code long} array
+ * @return a {@code Long} array, {@code null} if null array input
+ */
+ public static Long[] toObject(final long[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_LONG_OBJECT_ARRAY;
+ }
+ final Long[] result = new Long[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of primitive shorts to objects.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code short} array
+ * @return a {@code Short} array, {@code null} if null array input
+ */
+ public static Short[] toObject(final short[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_SHORT_OBJECT_ARRAY;
+ }
+ final Short[] result = new Short[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of object Booleans to primitives.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code Boolean} array, may be {@code null}
+ * @return a {@code boolean} array, {@code null} if null array input
+ * @throws NullPointerException if array content is {@code null}
+ */
+ public static boolean[] toPrimitive(final Boolean[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_BOOLEAN_ARRAY;
+ }
+ final boolean[] result = new boolean[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of object Bytes to primitives.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code Byte} array, may be {@code null}
+ * @return a {@code byte} array, {@code null} if null array input
+ * @throws NullPointerException if array content is {@code null}
+ */
+ public static byte[] toPrimitive(final Byte[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_BYTE_ARRAY;
+ }
+ final byte[] result = new byte[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of object Doubles to primitives.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code Double} array, may be {@code null}
+ * @return a {@code double} array, {@code null} if null array input
+ * @throws NullPointerException if array content is {@code null}
+ */
+ public static double[] toPrimitive(final Double[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_DOUBLE_ARRAY;
+ }
+ final double[] result = new double[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of object Floats to primitives.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code Float} array, may be {@code null}
+ * @return a {@code float} array, {@code null} if null array input
+ * @throws NullPointerException if array content is {@code null}
+ */
+ public static float[] toPrimitive(final Float[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_FLOAT_ARRAY;
+ }
+ final float[] result = new float[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of object Integers to primitives.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code Integer} array, may be {@code null}
+ * @return an {@code int} array, {@code null} if null array input
+ * @throws NullPointerException if array content is {@code null}
+ */
+ public static int[] toPrimitive(final Integer[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_INT_ARRAY;
+ }
+ final int[] result = new int[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of object Longs to primitives.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code Long} array, may be {@code null}
+ * @return a {@code long} array, {@code null} if null array input
+ * @throws NullPointerException if array content is {@code null}
+ */
+ public static long[] toPrimitive(final Long[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_LONG_ARRAY;
+ }
+ final long[] result = new long[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+
+ /**
+ * Converts an array of object Shorts to primitives.
+ *
+ * <p>This method returns {@code null} for a {@code null} input array.
+ *
+ * @param array a {@code Short} array, may be {@code null}
+ * @return a {@code byte} array, {@code null} if null array input
+ * @throws NullPointerException if array content is {@code null}
+ */
+ public static short[] toPrimitive(final Short[] array) {
+ if (array == null) {
+ return null;
+ } else if (array.length == 0) {
+ return EMPTY_SHORT_ARRAY;
+ }
+ final short[] result = new short[array.length];
+ for (int i = 0; i < array.length; i++) {
+ result[i] = array[i];
+ }
+ return result;
+ }
+}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/CloseableIterator.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/CloseableIterator.java
new file mode 100644
index 00000000..dbeba664
--- /dev/null
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/CloseableIterator.java
@@ -0,0 +1,191 @@
+/*
+ * 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.flink.table.store.utils;
+
+import javax.annotation.Nonnull;
+
+import java.util.ArrayDeque;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+import java.util.function.Consumer;
+
+import static java.util.Arrays.asList;
+
+/**
+ * This interface represents an {@link Iterator} that is also {@link AutoCloseable}. A typical
+ * use-case for this interface are iterators that are based on native-resources such as files,
+ * network, or database connections. Clients must call {@link #close()} after using the iterator.
+ *
+ * @param <T> the type of iterated elements.
+ */
+public interface CloseableIterator<T> extends Iterator<T>, AutoCloseable {
+
+ CloseableIterator<?> EMPTY_INSTANCE =
+ CloseableIterator.adapterForIterator(Collections.emptyIterator());
+
+ @Nonnull
+ static <T> CloseableIterator<T> adapterForIterator(@Nonnull Iterator<T> iterator) {
+ return adapterForIterator(iterator, () -> {});
+ }
+
+ static <T> CloseableIterator<T> adapterForIterator(
+ @Nonnull Iterator<T> iterator, AutoCloseable close) {
+ return new IteratorAdapter<>(iterator, close);
+ }
+
+ static <T> CloseableIterator<T> fromList(List<T> list, Consumer<T> closeNotConsumed) {
+ return new CloseableIterator<T>() {
+ private final Deque<T> stack = new ArrayDeque<>(list);
+
+ @Override
+ public boolean hasNext() {
+ return !stack.isEmpty();
+ }
+
+ @Override
+ public T next() {
+ return stack.poll();
+ }
+
+ @Override
+ public void close() throws Exception {
+ Exception exception = null;
+ for (T el : stack) {
+ try {
+ closeNotConsumed.accept(el);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ }
+ if (exception != null) {
+ throw exception;
+ }
+ }
+ };
+ }
+
+ static <T> CloseableIterator<T> flatten(CloseableIterator<T>... iterators) {
+ return new CloseableIterator<T>() {
+ private final Queue<CloseableIterator<T>> queue =
+ removeEmptyHead(new LinkedList<>(asList(iterators)));
+
+ private Queue<CloseableIterator<T>> removeEmptyHead(Queue<CloseableIterator<T>> queue) {
+ while (!queue.isEmpty() && !queue.peek().hasNext()) {
+ queue.poll();
+ }
+ return queue;
+ }
+
+ @Override
+ public boolean hasNext() {
+ removeEmptyHead(queue);
+ return !queue.isEmpty();
+ }
+
+ @Override
+ public T next() {
+ removeEmptyHead(queue);
+ return queue.peek().next();
+ }
+
+ @Override
+ public void close() throws Exception {
+ IOUtils.closeAll(iterators);
+ }
+ };
+ }
+
+ @SuppressWarnings("unchecked")
+ static <T> CloseableIterator<T> empty() {
+ return (CloseableIterator<T>) EMPTY_INSTANCE;
+ }
+
+ static <T> CloseableIterator<T> ofElements(Consumer<T> closeNotConsumed, T... elements) {
+ return fromList(asList(elements), closeNotConsumed);
+ }
+
+ static <E> CloseableIterator<E> ofElement(E element, Consumer<E> closeIfNotConsumed) {
+ return new CloseableIterator<E>() {
+ private boolean hasNext = true;
+
+ @Override
+ public boolean hasNext() {
+ return hasNext;
+ }
+
+ @Override
+ public E next() {
+ hasNext = false;
+ return element;
+ }
+
+ @Override
+ public void close() {
+ if (hasNext) {
+ closeIfNotConsumed.accept(element);
+ }
+ }
+ };
+ }
+
+ /**
+ * Adapter from {@link Iterator} to {@link CloseableIterator}. Does nothing on {@link #close()}.
+ *
+ * @param <E> the type of iterated elements.
+ */
+ final class IteratorAdapter<E> implements CloseableIterator<E> {
+
+ @Nonnull private final Iterator<E> delegate;
+ private final AutoCloseable close;
+
+ IteratorAdapter(@Nonnull Iterator<E> delegate, AutoCloseable close) {
+ this.delegate = delegate;
+ this.close = close;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return delegate.hasNext();
+ }
+
+ @Override
+ public E next() {
+ return delegate.next();
+ }
+
+ @Override
+ public void remove() {
+ delegate.remove();
+ }
+
+ @Override
+ public void forEachRemaining(Consumer<? super E> action) {
+ delegate.forEachRemaining(action);
+ }
+
+ @Override
+ public void close() throws Exception {
+ close.close();
+ }
+ }
+}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/ExceptionUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/ExceptionUtils.java
new file mode 100644
index 00000000..36654142
--- /dev/null
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/ExceptionUtils.java
@@ -0,0 +1,576 @@
+/*
+ * 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.
+ */
+
+//
+// The function "stringifyException" is based on source code from the Hadoop Project
+// (http://hadoop.apache.org/),
+// licensed by the Apache Software Foundation (ASF) under the Apache License, Version 2.0.
+// See the NOTICE file distributed with this work for additional information regarding copyright
+// ownership.
+//
+
+package org.apache.flink.table.store.utils;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.lang.reflect.Field;
+import java.util.Locale;
+import java.util.Optional;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+import java.util.function.Predicate;
+
+import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
+
+/** A collection of utility functions for dealing with exceptions and exception workflows. */
+public final class ExceptionUtils {
+
+ /** The stringified representation of a null exception reference. */
+ public static final String STRINGIFIED_NULL_EXCEPTION = "(null)";
+
+ /**
+ * Makes a string representation of the exception's stack trace, or "(null)", if the exception
+ * is null.
+ *
+ * <p>This method makes a best effort and never fails.
+ *
+ * @param e The exception to stringify.
+ * @return A string with exception name and call stack.
+ */
+ public static String stringifyException(final Throwable e) {
+ if (e == null) {
+ return STRINGIFIED_NULL_EXCEPTION;
+ }
+
+ try {
+ StringWriter stm = new StringWriter();
+ PrintWriter wrt = new PrintWriter(stm);
+ e.printStackTrace(wrt);
+ wrt.close();
+ return stm.toString();
+ } catch (Throwable t) {
+ return e.getClass().getName() + " (error while printing stack trace)";
+ }
+ }
+
+ /**
+ * Checks whether the given exception indicates a situation that may leave the JVM in a
+ * corrupted state, meaning a state where continued normal operation can only be guaranteed via
+ * clean process restart.
+ *
+ * <p>Currently considered fatal exceptions are Virtual Machine errors indicating that the JVM
+ * is corrupted, like {@link InternalError}, {@link UnknownError}, and {@link
+ * java.util.zip.ZipError} (a special case of InternalError). The {@link ThreadDeath} exception
+ * is also treated as a fatal error, because when a thread is forcefully stopped, there is a
+ * high chance that parts of the system are in an inconsistent state.
+ *
+ * @param t The exception to check.
+ * @return True, if the exception is considered fatal to the JVM, false otherwise.
+ */
+ public static boolean isJvmFatalError(Throwable t) {
+ return (t instanceof InternalError)
+ || (t instanceof UnknownError)
+ || (t instanceof ThreadDeath);
+ }
+
+ /**
+ * Checks whether the given exception indicates a situation that may leave the JVM in a
+ * corrupted state, or an out-of-memory error.
+ *
+ * <p>See {@link ExceptionUtils#isJvmFatalError(Throwable)} for a list of fatal JVM errors. This
+ * method additionally classifies the {@link OutOfMemoryError} as fatal, because it may occur in
+ * any thread (not the one that allocated the majority of the memory) and thus is often not
+ * recoverable by destroying the particular thread that threw the exception.
+ *
+ * @param t The exception to check.
+ * @return True, if the exception is fatal to the JVM or and OutOfMemoryError, false otherwise.
+ */
+ public static boolean isJvmFatalOrOutOfMemoryError(Throwable t) {
+ return isJvmFatalError(t) || t instanceof OutOfMemoryError;
+ }
+
+ /**
+ * Tries to enrich OutOfMemoryErrors being part of the passed root Throwable's cause tree.
+ *
+ * <p>This method improves error messages for direct and metaspace {@link OutOfMemoryError}. It
+ * adds description about the possible causes and ways of resolution.
+ *
+ * @param root The Throwable of which the cause tree shall be traversed.
+ * @param jvmMetaspaceOomNewErrorMessage The message being used for JVM metaspace-related
+ * OutOfMemoryErrors. Passing <code>null</code> will disable handling this class of error.
+ * @param jvmDirectOomNewErrorMessage The message being used for direct memory-related
+ * OutOfMemoryErrors. Passing <code>null</code> will disable handling this class of error.
+ * @param jvmHeapSpaceOomNewErrorMessage The message being used for Heap space-related
+ * OutOfMemoryErrors. Passing <code>null</code> will disable handling this class of error.
+ */
+ public static void tryEnrichOutOfMemoryError(
+ @Nullable Throwable root,
+ @Nullable String jvmMetaspaceOomNewErrorMessage,
+ @Nullable String jvmDirectOomNewErrorMessage,
+ @Nullable String jvmHeapSpaceOomNewErrorMessage) {
+ updateDetailMessage(
+ root,
+ t -> {
+ if (isMetaspaceOutOfMemoryError(t)) {
+ return jvmMetaspaceOomNewErrorMessage;
+ } else if (isDirectOutOfMemoryError(t)) {
+ return jvmDirectOomNewErrorMessage;
+ } else if (isHeapSpaceOutOfMemoryError(t)) {
+ return jvmHeapSpaceOomNewErrorMessage;
+ }
+
+ return null;
+ });
+ }
+
+ /**
+ * Updates error messages of Throwables appearing in the cause tree of the passed root
+ * Throwable. The passed Function is applied on each Throwable of the cause tree. Returning a
+ * String will cause the detailMessage of the corresponding Throwable to be updated. Returning
+ * <code>null</code>, instead, won't trigger any detailMessage update on that Throwable.
+ *
+ * @param root The Throwable whose cause tree shall be traversed.
+ * @param throwableToMessage The Function based on which the new messages are generated. The
+ * function implementation should return the new message. Returning <code>null</code>, in
+ * contrast, will result in not updating the message for the corresponding Throwable.
+ */
+ public static void updateDetailMessage(
+ @Nullable Throwable root, @Nullable Function<Throwable, String> throwableToMessage) {
+ if (throwableToMessage == null) {
+ return;
+ }
+
+ Throwable it = root;
+ while (it != null) {
+ String newMessage = throwableToMessage.apply(it);
+ if (newMessage != null) {
+ updateDetailMessageOfThrowable(it, newMessage);
+ }
+
+ it = it.getCause();
+ }
+ }
+
+ private static void updateDetailMessageOfThrowable(
+ Throwable throwable, String newDetailMessage) {
+ Field field;
+ try {
+ field = Throwable.class.getDeclaredField("detailMessage");
+ } catch (NoSuchFieldException e) {
+ throw new IllegalStateException(
+ "The JDK Throwable contains a detailMessage member. The Throwable class provided on the classpath does not which is why this exception appears.",
+ e);
+ }
+
+ field.setAccessible(true);
+ try {
+ field.set(throwable, newDetailMessage);
+ } catch (IllegalAccessException e) {
+ throw new IllegalStateException(
+ "The JDK Throwable contains a private detailMessage member that should be accessible through reflection. This is not the case for the Throwable class provided on the classpath.",
+ e);
+ }
+ }
+
+ /**
+ * Checks whether the given exception indicates a JVM metaspace out-of-memory error.
+ *
+ * @param t The exception to check.
+ * @return True, if the exception is the metaspace {@link OutOfMemoryError}, false otherwise.
+ */
+ public static boolean isMetaspaceOutOfMemoryError(@Nullable Throwable t) {
+ return isOutOfMemoryErrorWithMessageContaining(t, "Metaspace");
+ }
+
+ /**
+ * Checks whether the given exception indicates a JVM direct out-of-memory error.
+ *
+ * @param t The exception to check.
+ * @return True, if the exception is the direct {@link OutOfMemoryError}, false otherwise.
+ */
+ public static boolean isDirectOutOfMemoryError(@Nullable Throwable t) {
+ return isOutOfMemoryErrorWithMessageContaining(t, "Direct buffer memory");
+ }
+
+ public static boolean isHeapSpaceOutOfMemoryError(@Nullable Throwable t) {
+ return isOutOfMemoryErrorWithMessageContaining(t, "Java heap space");
+ }
+
+ private static boolean isOutOfMemoryErrorWithMessageContaining(
+ @Nullable Throwable t, String infix) {
+ // the exact matching of the class is checked to avoid matching any custom subclasses of
+ // OutOfMemoryError
+ // as we are interested in the original exceptions, generated by JVM.
+ return isOutOfMemoryError(t)
+ && t.getMessage() != null
+ && t.getMessage().toLowerCase(Locale.ROOT).contains(infix.toLowerCase(Locale.ROOT));
+ }
+
+ private static boolean isOutOfMemoryError(@Nullable Throwable t) {
+ return t != null && t.getClass() == OutOfMemoryError.class;
+ }
+
+ /**
+ * Rethrows the given {@code Throwable}, if it represents an error that is fatal to the JVM. See
+ * {@link ExceptionUtils#isJvmFatalError(Throwable)} for a definition of fatal errors.
+ *
+ * @param t The Throwable to check and rethrow.
+ */
+ public static void rethrowIfFatalError(Throwable t) {
+ if (isJvmFatalError(t)) {
+ throw (Error) t;
+ }
+ }
+
+ /**
+ * Rethrows the given {@code Throwable}, if it represents an error that is fatal to the JVM or
+ * an out-of-memory error. See {@link ExceptionUtils#isJvmFatalError(Throwable)} for a
+ * definition of fatal errors.
+ *
+ * @param t The Throwable to check and rethrow.
+ */
+ public static void rethrowIfFatalErrorOrOOM(Throwable t) {
+ if (isJvmFatalError(t) || t instanceof OutOfMemoryError) {
+ throw (Error) t;
+ }
+ }
+
+ /**
+ * Adds a new exception as a {@link Throwable#addSuppressed(Throwable) suppressed exception} to
+ * a prior exception, or returns the new exception, if no prior exception exists.
+ *
+ * <pre>{@code
+ * public void closeAllThings() throws Exception {
+ * Exception ex = null;
+ * try {
+ * component.shutdown();
+ * } catch (Exception e) {
+ * ex = firstOrSuppressed(e, ex);
+ * }
+ * try {
+ * anotherComponent.stop();
+ * } catch (Exception e) {
+ * ex = firstOrSuppressed(e, ex);
+ * }
+ * try {
+ * lastComponent.shutdown();
+ * } catch (Exception e) {
+ * ex = firstOrSuppressed(e, ex);
+ * }
+ *
+ * if (ex != null) {
+ * throw ex;
+ * }
+ * }
+ * }</pre>
+ *
+ * @param newException The newly occurred exception
+ * @param previous The previously occurred exception, possibly null.
+ * @return The new exception, if no previous exception exists, or the previous exception with
+ * the new exception in the list of suppressed exceptions.
+ */
+ public static <T extends Throwable> T firstOrSuppressed(T newException, @Nullable T previous) {
+ checkNotNull(newException, "newException");
+
+ if (previous == null || previous == newException) {
+ return newException;
+ } else {
+ previous.addSuppressed(newException);
+ return previous;
+ }
+ }
+
+ /**
+ * Throws the given {@code Throwable} in scenarios where the signatures do not allow you to
+ * throw an arbitrary Throwable. Errors and RuntimeExceptions are thrown directly, other
+ * exceptions are packed into runtime exceptions
+ *
+ * @param t The throwable to be thrown.
+ */
+ public static void rethrow(Throwable t) {
+ if (t instanceof Error) {
+ throw (Error) t;
+ } else if (t instanceof RuntimeException) {
+ throw (RuntimeException) t;
+ } else {
+ throw new RuntimeException(t);
+ }
+ }
+
+ /**
+ * Throws the given {@code Throwable} in scenarios where the signatures do not allow you to
+ * throw an arbitrary Throwable. Errors and RuntimeExceptions are thrown directly, other
+ * exceptions are packed into a parent RuntimeException.
+ *
+ * @param t The throwable to be thrown.
+ * @param parentMessage The message for the parent RuntimeException, if one is needed.
+ */
+ public static void rethrow(Throwable t, String parentMessage) {
+ if (t instanceof Error) {
+ throw (Error) t;
+ } else if (t instanceof RuntimeException) {
+ throw (RuntimeException) t;
+ } else {
+ throw new RuntimeException(parentMessage, t);
+ }
+ }
+
+ /**
+ * Throws the given {@code Throwable} in scenarios where the signatures do allow to throw a
+ * Exception. Errors and Exceptions are thrown directly, other "exotic" subclasses of Throwable
+ * are wrapped in an Exception.
+ *
+ * @param t The throwable to be thrown.
+ * @param parentMessage The message for the parent Exception, if one is needed.
+ */
+ public static void rethrowException(Throwable t, String parentMessage) throws Exception {
+ if (t instanceof Error) {
+ throw (Error) t;
+ } else if (t instanceof Exception) {
+ throw (Exception) t;
+ } else {
+ throw new Exception(parentMessage, t);
+ }
+ }
+
+ /**
+ * Throws the given {@code Throwable} in scenarios where the signatures do allow to throw a
+ * Exception. Errors and Exceptions are thrown directly, other "exotic" subclasses of Throwable
+ * are wrapped in an Exception.
+ *
+ * @param t The throwable to be thrown.
+ */
+ public static void rethrowException(Throwable t) throws Exception {
+ if (t instanceof Error) {
+ throw (Error) t;
+ } else if (t instanceof Exception) {
+ throw (Exception) t;
+ } else {
+ throw new Exception(t.getMessage(), t);
+ }
+ }
+
+ /**
+ * Tries to throw the given exception if not null.
+ *
+ * @param e exception to throw if not null.
+ * @throws Exception
+ */
+ public static void tryRethrowException(@Nullable Exception e) throws Exception {
+ if (e != null) {
+ throw e;
+ }
+ }
+
+ /**
+ * Tries to throw the given {@code Throwable} in scenarios where the signatures allows only
+ * IOExceptions (and RuntimeException and Error). Throws this exception directly, if it is an
+ * IOException, a RuntimeException, or an Error. Otherwise does nothing.
+ *
+ * @param t The Throwable to be thrown.
+ */
+ public static void tryRethrowIOException(Throwable t) throws IOException {
+ if (t instanceof IOException) {
+ throw (IOException) t;
+ } else if (t instanceof RuntimeException) {
+ throw (RuntimeException) t;
+ } else if (t instanceof Error) {
+ throw (Error) t;
+ }
+ }
+
+ /**
+ * Re-throws the given {@code Throwable} in scenarios where the signatures allows only
+ * IOExceptions (and RuntimeException and Error).
+ *
+ * <p>Throws this exception directly, if it is an IOException, a RuntimeException, or an Error.
+ * Otherwise it wraps it in an IOException and throws it.
+ *
+ * @param t The Throwable to be thrown.
+ */
+ public static void rethrowIOException(Throwable t) throws IOException {
+ if (t instanceof IOException) {
+ throw (IOException) t;
+ } else if (t instanceof RuntimeException) {
+ throw (RuntimeException) t;
+ } else if (t instanceof Error) {
+ throw (Error) t;
+ } else {
+ throw new IOException(t.getMessage(), t);
+ }
+ }
+
+ /**
+ * Checks whether a throwable chain contains a specific type of exception and returns it.
+ *
+ * @param throwable the throwable chain to check.
+ * @param searchType the type of exception to search for in the chain.
+ * @return Optional throwable of the requested type if available, otherwise empty
+ */
+ public static <T extends Throwable> Optional<T> findThrowable(
+ Throwable throwable, Class<T> searchType) {
+ if (throwable == null || searchType == null) {
+ return Optional.empty();
+ }
+
+ Throwable t = throwable;
+ while (t != null) {
+ if (searchType.isAssignableFrom(t.getClass())) {
+ return Optional.of(searchType.cast(t));
+ } else {
+ t = t.getCause();
+ }
+ }
+
+ return Optional.empty();
+ }
+
+ /**
+ * The same as {@link #findThrowable(Throwable, Class)}, but rethrows original exception if the
+ * expected exception was not found.
+ */
+ public static <T extends Throwable> void assertThrowable(
+ Throwable throwable, Class<T> searchType) throws T {
+ if (!findThrowable(throwable, searchType).isPresent()) {
+ throw (T) throwable;
+ }
+ }
+
+ /**
+ * Checks whether a throwable chain contains an exception matching a predicate and returns it.
+ *
+ * @param throwable the throwable chain to check.
+ * @param predicate the predicate of the exception to search for in the chain.
+ * @return Optional throwable of the requested type if available, otherwise empty
+ */
+ public static Optional<Throwable> findThrowable(
+ Throwable throwable, Predicate<Throwable> predicate) {
+ if (throwable == null || predicate == null) {
+ return Optional.empty();
+ }
+
+ Throwable t = throwable;
+ while (t != null) {
+ if (predicate.test(t)) {
+ return Optional.of(t);
+ } else {
+ t = t.getCause();
+ }
+ }
+
+ return Optional.empty();
+ }
+
+ /**
+ * Checks whether a throwable chain contains a specific error message and returns the
+ * corresponding throwable.
+ *
+ * @param throwable the throwable chain to check.
+ * @param searchMessage the error message to search for in the chain.
+ * @return Optional throwable containing the search message if available, otherwise empty
+ */
+ public static Optional<Throwable> findThrowableWithMessage(
+ Throwable throwable, String searchMessage) {
+ if (throwable == null || searchMessage == null) {
+ return Optional.empty();
+ }
+
+ Throwable t = throwable;
+ while (t != null) {
+ if (t.getMessage() != null && t.getMessage().contains(searchMessage)) {
+ return Optional.of(t);
+ } else {
+ t = t.getCause();
+ }
+ }
+
+ return Optional.empty();
+ }
+
+ /**
+ * The same as {@link #findThrowableWithMessage(Throwable, String)}, but rethrows original
+ * exception if the expected exception was not found.
+ */
+ public static <T extends Throwable> void assertThrowableWithMessage(
+ Throwable throwable, String searchMessage) throws T {
+ if (!findThrowableWithMessage(throwable, searchMessage).isPresent()) {
+ throw (T) throwable;
+ }
+ }
+
+ /**
+ * Unpacks an {@link ExecutionException} and returns its cause. Otherwise the given Throwable is
+ * returned.
+ *
+ * @param throwable to unpack if it is an ExecutionException
+ * @return Cause of ExecutionException or given Throwable
+ */
+ public static Throwable stripExecutionException(Throwable throwable) {
+ return stripException(throwable, ExecutionException.class);
+ }
+
+ /**
+ * Unpacks an {@link CompletionException} and returns its cause. Otherwise the given Throwable
+ * is returned.
+ *
+ * @param throwable to unpack if it is an CompletionException
+ * @return Cause of CompletionException or given Throwable
+ */
+ public static Throwable stripCompletionException(Throwable throwable) {
+ return stripException(throwable, CompletionException.class);
+ }
+
+ /**
+ * Unpacks an specified exception and returns its cause. Otherwise the given {@link Throwable}
+ * is returned.
+ *
+ * @param throwableToStrip to strip
+ * @param typeToStrip type to strip
+ * @return Unpacked cause or given Throwable if not packed
+ */
+ public static Throwable stripException(
+ Throwable throwableToStrip, Class<? extends Throwable> typeToStrip) {
+ while (typeToStrip.isAssignableFrom(throwableToStrip.getClass())
+ && throwableToStrip.getCause() != null) {
+ throwableToStrip = throwableToStrip.getCause();
+ }
+
+ return throwableToStrip;
+ }
+
+ /**
+ * Checks whether the given exception is a {@link InterruptedException} and sets the interrupted
+ * flag accordingly.
+ *
+ * @param e to check whether it is an {@link InterruptedException}
+ */
+ public static void checkInterrupted(Throwable e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /** Private constructor to prevent instantiation. */
+ private ExceptionUtils() {}
+}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/FunctionWithException.java
similarity index 50%
copy from flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java
copy to flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/FunctionWithException.java
index 7a153d54..26f309ba 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/FunctionWithException.java
@@ -16,25 +16,24 @@
* limitations under the License.
*/
-package org.apache.flink.table.store.file.predicate;
+package org.apache.flink.table.store.utils;
-import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArrayComparator;
-import org.apache.flink.table.store.types.DataType;
-
-/** Utils for comparator. */
-public class CompareUtils {
- private CompareUtils() {}
-
- private static final BytePrimitiveArrayComparator BINARY_COMPARATOR =
- new BytePrimitiveArrayComparator(true);
+/**
+ * A functional interface for a {@link java.util.function.Function} that may throw exceptions.
+ *
+ * @param <T> The type of the argument to the function.
+ * @param <R> The type of the result of the supplier.
+ * @param <E> The type of Exceptions thrown by this function.
+ */
+@FunctionalInterface
+public interface FunctionWithException<T, R, E extends Throwable> {
- public static int compareLiteral(DataType type, Object v1, Object v2) {
- if (v1 instanceof Comparable) {
- return ((Comparable<Object>) v1).compareTo(v2);
- } else if (v1 instanceof byte[]) {
- return BINARY_COMPARATOR.compare((byte[]) v1, (byte[]) v2);
- } else {
- throw new RuntimeException("Unsupported type: " + type);
- }
- }
+ /**
+ * Calls this function.
+ *
+ * @param value The argument to the function.
+ * @return The result of thus supplier.
+ * @throws E This function may throw an exception.
+ */
+ R apply(T value) throws E;
}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/IOUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/IOUtils.java
index 031eee39..188c0fc3 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/IOUtils.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/IOUtils.java
@@ -18,13 +18,13 @@
package org.apache.flink.table.store.utils;
-import org.apache.flink.util.ExceptionUtils;
-
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.PrintStream;
+import static java.util.Arrays.asList;
+
/** An utility class for I/O related functionality. */
public final class IOUtils {
@@ -112,6 +112,11 @@ public final class IOUtils {
// Silent I/O cleanup / closing
// ------------------------------------------------------------------------
+ /** @see #closeAll(Iterable) */
+ public static void closeAll(AutoCloseable... closeables) throws Exception {
+ closeAll(asList(closeables));
+ }
+
/**
* Closes all {@link AutoCloseable} objects in the parameter, suppressing exceptions. Exception
* will be emitted after calling close() on every object.
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/LongCounter.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/LongCounter.java
new file mode 100644
index 00000000..77a3d958
--- /dev/null
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/LongCounter.java
@@ -0,0 +1,63 @@
+/*
+ * 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.flink.table.store.utils;
+
+import java.io.Serializable;
+
+/** An counter that sums up {@code long} values. */
+public class LongCounter implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private long value;
+
+ public LongCounter() {
+ this(0);
+ }
+
+ public LongCounter(long value) {
+ this.value = value;
+ }
+
+ /** Consider using {@link #add(long)} instead for primitive long values. */
+ public void add(Long value) {
+ this.value += value;
+ }
+
+ public Long getValue() {
+ return this.value;
+ }
+
+ public void merge(LongCounter other) {
+ this.value += other.getValue();
+ }
+
+ public void reset() {
+ this.value = 0;
+ }
+
+ public void add(long value) {
+ this.value += value;
+ }
+
+ @Override
+ public String toString() {
+ return "LongCounter " + this.value;
+ }
+}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/Pair.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/Pair.java
new file mode 100644
index 00000000..d7423ab9
--- /dev/null
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/Pair.java
@@ -0,0 +1,98 @@
+/*
+ * 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.flink.table.store.utils;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.Objects;
+
+/** Pair contains left and right. */
+public final class Pair<L, R> implements Map.Entry<L, R>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Left object. */
+ private L left;
+
+ /** Right object. */
+ private R right;
+
+ public static <L, R> Pair<L, R> of(L left, R right) {
+ return new Pair<>(left, right);
+ }
+
+ private Pair(final L left, final R right) {
+ this.left = left;
+ this.right = right;
+ }
+
+ public L getLeft() {
+ return left;
+ }
+
+ public R getRight() {
+ return right;
+ }
+
+ public void setLeft(L left) {
+ this.left = left;
+ }
+
+ public R setRight(R right) {
+ R previous = this.right;
+ this.right = right;
+ return previous;
+ }
+
+ @Override
+ public L getKey() {
+ return getLeft();
+ }
+
+ @Override
+ public R getValue() {
+ return getRight();
+ }
+
+ @Override
+ public R setValue(R value) {
+ return setRight(value);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Pair<?, ?> pair = (Pair<?, ?>) o;
+ return Objects.equals(left, pair.left) && Objects.equals(right, pair.right);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(left, right);
+ }
+
+ @Override
+ public String toString() {
+ return "(" + getLeft() + ',' + getRight() + ')';
+ }
+}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/SerializableFunction.java
similarity index 50%
copy from flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java
copy to flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/SerializableFunction.java
index 7a153d54..f3edb145 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/CompareUtils.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/SerializableFunction.java
@@ -16,25 +16,11 @@
* limitations under the License.
*/
-package org.apache.flink.table.store.file.predicate;
+package org.apache.flink.table.store.utils;
-import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArrayComparator;
-import org.apache.flink.table.store.types.DataType;
+import java.io.Serializable;
+import java.util.function.Function;
-/** Utils for comparator. */
-public class CompareUtils {
- private CompareUtils() {}
-
- private static final BytePrimitiveArrayComparator BINARY_COMPARATOR =
- new BytePrimitiveArrayComparator(true);
-
- public static int compareLiteral(DataType type, Object v1, Object v2) {
- if (v1 instanceof Comparable) {
- return ((Comparable<Object>) v1).compareTo(v2);
- } else if (v1 instanceof byte[]) {
- return BINARY_COMPARATOR.compare((byte[]) v1, (byte[]) v2);
- } else {
- throw new RuntimeException("Unsupported type: " + type);
- }
- }
-}
+/** A {@link Function} that is also {@link Serializable}. */
+@FunctionalInterface
+public interface SerializableFunction<T, R> extends Function<T, R>, Serializable {}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/StringUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/StringUtils.java
index 5cab3b7d..47b25e80 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/StringUtils.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/StringUtils.java
@@ -21,26 +21,32 @@ package org.apache.flink.table.store.utils;
import org.apache.flink.table.store.data.BinaryString;
import org.apache.flink.table.store.memory.MemorySegmentUtils;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.List;
import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
import static org.apache.flink.table.store.data.BinaryString.fromBytes;
+import static org.apache.flink.table.store.utils.Preconditions.checkArgument;
+import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
/**
* Utils for {@link BinaryString} and utility class to convert objects into strings in vice-versa.
*/
public class StringUtils {
- /**
- * An empty string array. There are just too many places where one needs an empty string array
- * and wants to save some object allocation.
- */
- public static final String[] EMPTY_STRING_ARRAY = new String[0];
-
private static final char[] HEX_CHARS = {
'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'
};
+ /**
+ * Represents a failed index search.
+ *
+ * @since 2.1
+ */
+ public static final int INDEX_NOT_FOUND = -1;
+
/**
* Concatenates input strings together into a single string. Returns NULL if any argument is
* NULL.
@@ -173,4 +179,281 @@ public class StringUtils {
}
return new String(data);
}
+
+ /**
+ * Returns a string consisting of a specific number of concatenated copies of an input string.
+ * For example, {@code repeat("hey", 3)} returns the string {@code "heyheyhey"}.
+ *
+ * @param string any non-null string
+ * @param count the number of times to repeat it; a nonnegative integer
+ * @return a string containing {@code string} repeated {@code count} times (the empty string if
+ * {@code count} is zero)
+ * @throws IllegalArgumentException if {@code count} is negative
+ */
+ public static String repeat(String string, int count) {
+ checkNotNull(string); // eager for GWT.
+
+ if (count <= 1) {
+ checkArgument(count >= 0, "invalid count: %s", count);
+ return (count == 0) ? "" : string;
+ }
+
+ // IF YOU MODIFY THE CODE HERE, you must update StringsRepeatBenchmark
+ final int len = string.length();
+ final long longSize = (long) len * (long) count;
+ final int size = (int) longSize;
+ if (size != longSize) {
+ throw new ArrayIndexOutOfBoundsException("Required array size too large: " + longSize);
+ }
+
+ final char[] array = new char[size];
+ string.getChars(0, len, array, 0);
+ int n;
+ for (n = len; n < size - n; n <<= 1) {
+ System.arraycopy(array, 0, array, n, n);
+ }
+ System.arraycopy(array, 0, array, n, size - n);
+ return new String(array);
+ }
+
+ /**
+ * Replaces all occurrences of a String within another String.
+ *
+ * <p>A {@code null} reference passed to this method is a no-op.
+ *
+ * <pre>
+ * StringUtils.replace(null, *, *) = null
+ * StringUtils.replace("", *, *) = ""
+ * StringUtils.replace("any", null, *) = "any"
+ * StringUtils.replace("any", *, null) = "any"
+ * StringUtils.replace("any", "", *) = "any"
+ * StringUtils.replace("aba", "a", null) = "aba"
+ * StringUtils.replace("aba", "a", "") = "b"
+ * StringUtils.replace("aba", "a", "z") = "zbz"
+ * </pre>
+ *
+ * @see #replace(String text, String searchString, String replacement, int max)
+ * @param text text to search and replace in, may be null
+ * @param searchString the String to search for, may be null
+ * @param replacement the String to replace it with, may be null
+ * @return the text with any replacements processed, {@code null} if null String input
+ */
+ public static String replace(
+ final String text, final String searchString, final String replacement) {
+ return replace(text, searchString, replacement, -1);
+ }
+
+ /**
+ * Replaces a String with another String inside a larger String, for the first {@code max}
+ * values of the search String.
+ *
+ * <p>A {@code null} reference passed to this method is a no-op.
+ *
+ * <pre>
+ * StringUtils.replace(null, *, *, *) = null
+ * StringUtils.replace("", *, *, *) = ""
+ * StringUtils.replace("any", null, *, *) = "any"
+ * StringUtils.replace("any", *, null, *) = "any"
+ * StringUtils.replace("any", "", *, *) = "any"
+ * StringUtils.replace("any", *, *, 0) = "any"
+ * StringUtils.replace("abaa", "a", null, -1) = "abaa"
+ * StringUtils.replace("abaa", "a", "", -1) = "b"
+ * StringUtils.replace("abaa", "a", "z", 0) = "abaa"
+ * StringUtils.replace("abaa", "a", "z", 1) = "zbaa"
+ * StringUtils.replace("abaa", "a", "z", 2) = "zbza"
+ * StringUtils.replace("abaa", "a", "z", -1) = "zbzz"
+ * </pre>
+ *
+ * @param text text to search and replace in, may be null
+ * @param searchString the String to search for, may be null
+ * @param replacement the String to replace it with, may be null
+ * @param max maximum number of values to replace, or {@code -1} if no maximum
+ * @return the text with any replacements processed, {@code null} if null String input
+ */
+ public static String replace(
+ final String text, final String searchString, final String replacement, int max) {
+ if (isEmpty(text) || isEmpty(searchString) || replacement == null || max == 0) {
+ return text;
+ }
+ int start = 0;
+ int end = text.indexOf(searchString, start);
+ if (end == INDEX_NOT_FOUND) {
+ return text;
+ }
+ final int replLength = searchString.length();
+ int increase = replacement.length() - replLength;
+ increase = increase < 0 ? 0 : increase;
+ increase *= max < 0 ? 16 : max > 64 ? 64 : max;
+ final StringBuilder buf = new StringBuilder(text.length() + increase);
+ while (end != INDEX_NOT_FOUND) {
+ buf.append(text.substring(start, end)).append(replacement);
+ start = end + replLength;
+ if (--max == 0) {
+ break;
+ }
+ end = text.indexOf(searchString, start);
+ }
+ buf.append(text.substring(start));
+ return buf.toString();
+ }
+
+ /**
+ * Checks if a CharSequence is empty ("") or null.
+ *
+ * <pre>
+ * StringUtils.isEmpty(null) = true
+ * StringUtils.isEmpty("") = true
+ * StringUtils.isEmpty(" ") = false
+ * StringUtils.isEmpty("bob") = false
+ * StringUtils.isEmpty(" bob ") = false
+ * </pre>
+ *
+ * <p>NOTE: This method changed in Lang version 2.0. It no longer trims the CharSequence. That
+ * functionality is available in isBlank().
+ *
+ * @param cs the CharSequence to check, may be null
+ * @return {@code true} if the CharSequence is empty or null
+ * @since 3.0 Changed signature from isEmpty(String) to isEmpty(CharSequence)
+ */
+ public static boolean isEmpty(final CharSequence cs) {
+ return cs == null || cs.length() == 0;
+ }
+
+ public static String randomNumericString(int len) {
+ StringBuilder builder = new StringBuilder();
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+ for (int i = 0; i < len; i++) {
+ builder.append(rnd.nextInt(10));
+ }
+ return builder.toString();
+ }
+
+ /**
+ * Splits the provided text into an array, separators specified. This is an alternative to using
+ * StringTokenizer.
+ *
+ * <p>The separator is not included in the returned String array. Adjacent separators are
+ * treated as one separator. For more control over the split use the StrTokenizer class.
+ *
+ * <p>A {@code null} input String returns {@code null}. A {@code null} separatorChars splits on
+ * whitespace.
+ *
+ * <pre>
+ * StringUtils.split(null, *) = null
+ * StringUtils.split("", *) = []
+ * StringUtils.split("abc def", null) = ["abc", "def"]
+ * StringUtils.split("abc def", " ") = ["abc", "def"]
+ * StringUtils.split("abc def", " ") = ["abc", "def"]
+ * StringUtils.split("ab:cd:ef", ":") = ["ab", "cd", "ef"]
+ * </pre>
+ *
+ * @param str the String to parse, may be null
+ * @param separatorChars the characters used as the delimiters, {@code null} splits on
+ * whitespace
+ * @return an array of parsed Strings, {@code null} if null String input
+ */
+ public static String[] split(final String str, final String separatorChars) {
+ return splitWorker(str, separatorChars, -1, false);
+ }
+
+ /**
+ * Performs the logic for the {@code split} and {@code splitPreserveAllTokens} methods that
+ * return a maximum array length.
+ *
+ * @param str the String to parse, may be {@code null}
+ * @param separatorChars the separate character
+ * @param max the maximum number of elements to include in the array. A zero or negative value
+ * implies no limit.
+ * @param preserveAllTokens if {@code true}, adjacent separators are treated as empty token
+ * separators; if {@code false}, adjacent separators are treated as one separator.
+ * @return an array of parsed Strings, {@code null} if null String input
+ */
+ private static String[] splitWorker(
+ final String str,
+ final String separatorChars,
+ final int max,
+ final boolean preserveAllTokens) {
+ // Performance tuned for 2.0 (JDK1.4)
+ // Direct code is quicker than StringTokenizer.
+ // Also, StringTokenizer uses isSpace() not isWhitespace()
+
+ if (str == null) {
+ return null;
+ }
+ final int len = str.length();
+ if (len == 0) {
+ return ArrayUtils.EMPTY_STRING_ARRAY;
+ }
+ final List<String> list = new ArrayList<String>();
+ int sizePlus1 = 1;
+ int i = 0, start = 0;
+ boolean match = false;
+ boolean lastMatch = false;
+ if (separatorChars == null) {
+ // Null separator means use whitespace
+ while (i < len) {
+ if (Character.isWhitespace(str.charAt(i))) {
+ if (match || preserveAllTokens) {
+ lastMatch = true;
+ if (sizePlus1++ == max) {
+ i = len;
+ lastMatch = false;
+ }
+ list.add(str.substring(start, i));
+ match = false;
+ }
+ start = ++i;
+ continue;
+ }
+ lastMatch = false;
+ match = true;
+ i++;
+ }
+ } else if (separatorChars.length() == 1) {
+ // Optimise 1 character case
+ final char sep = separatorChars.charAt(0);
+ while (i < len) {
+ if (str.charAt(i) == sep) {
+ if (match || preserveAllTokens) {
+ lastMatch = true;
+ if (sizePlus1++ == max) {
+ i = len;
+ lastMatch = false;
+ }
+ list.add(str.substring(start, i));
+ match = false;
+ }
+ start = ++i;
+ continue;
+ }
+ lastMatch = false;
+ match = true;
+ i++;
+ }
+ } else {
+ // standard case
+ while (i < len) {
+ if (separatorChars.indexOf(str.charAt(i)) >= 0) {
+ if (match || preserveAllTokens) {
+ lastMatch = true;
+ if (sizePlus1++ == max) {
+ i = len;
+ lastMatch = false;
+ }
+ list.add(str.substring(start, i));
+ match = false;
+ }
+ start = ++i;
+ continue;
+ }
+ lastMatch = false;
+ match = true;
+ i++;
+ }
+ }
+ if (match || preserveAllTokens && lastMatch) {
+ list.add(str.substring(start, i));
+ }
+ return list.toArray(new String[list.size()]);
+ }
}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/TimeUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/TimeUtils.java
index 3256de5e..63e73170 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/TimeUtils.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/TimeUtils.java
@@ -18,8 +18,6 @@
package org.apache.flink.table.store.utils;
-import org.apache.flink.api.common.time.Time;
-
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Arrays;
@@ -233,16 +231,6 @@ public class TimeUtils {
}
}
- /**
- * Translates {@link Time} to {@link Duration}.
- *
- * @param time time to transform into duration
- * @return duration equal to the given time
- */
- public static Duration toDuration(Time time) {
- return Duration.of(time.getSize(), toChronoUnit(time.getUnit()));
- }
-
private static ChronoUnit toChronoUnit(java.util.concurrent.TimeUnit timeUnit) {
switch (timeUnit) {
case NANOSECONDS:
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/CodeRewriterTestBase.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/CodeRewriterTestBase.java
index fb4d2d9c..d5d84ca7 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/CodeRewriterTestBase.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/CodeRewriterTestBase.java
@@ -17,7 +17,7 @@
package org.apache.flink.table.store.codegen.codesplit;
-import org.apache.flink.util.FileUtils;
+import org.apache.flink.table.store.utils.FileIOUtils;
import java.io.File;
import java.util.function.Function;
@@ -39,7 +39,7 @@ abstract class CodeRewriterTestBase<R extends CodeRewriter> {
R runTest(String filename) {
try {
String code =
- FileUtils.readFileUtf8(
+ FileIOUtils.readFileUtf8(
new File(
CodeRewriterTestBase.class
.getClassLoader()
@@ -51,7 +51,7 @@ abstract class CodeRewriterTestBase<R extends CodeRewriter> {
+ ".java")
.toURI()));
String expected =
- FileUtils.readFileUtf8(
+ FileIOUtils.readFileUtf8(
new File(
CodeRewriterTestBase.class
.getClassLoader()
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/CodeSplitTestUtil.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/CodeSplitTestUtil.java
index f989d3ae..a46b216e 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/CodeSplitTestUtil.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/CodeSplitTestUtil.java
@@ -17,10 +17,10 @@
package org.apache.flink.table.store.codegen.codesplit;
-import org.apache.flink.util.FileUtils;
-import org.apache.flink.util.StringUtils;
+import org.apache.flink.table.store.utils.FileIOUtils;
+import org.apache.flink.table.store.utils.Pair;
+import org.apache.flink.table.store.utils.StringUtils;
-import org.apache.commons.lang3.tuple.Pair;
import org.codehaus.commons.compiler.CompileException;
import org.codehaus.janino.SimpleCompiler;
import org.junit.jupiter.api.Assertions;
@@ -34,7 +34,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;
-import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
import static org.assertj.core.api.Assertions.assertThat;
/** Util class for code split tests. */
@@ -70,7 +70,7 @@ public final class CodeSplitTestUtil {
public static String readResource(String resourcePath) {
try {
- return FileUtils.readFileUtf8(
+ return FileIOUtils.readFileUtf8(
new File(
BlockStatementGrouperTest.class
.getClassLoader()
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/JavaCodeSplitterTest.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/JavaCodeSplitterTest.java
index 1d8c98f4..f765ad18 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/JavaCodeSplitterTest.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/codegen/codesplit/JavaCodeSplitterTest.java
@@ -17,7 +17,7 @@
package org.apache.flink.table.store.codegen.codesplit;
-import org.apache.flink.util.FileUtils;
+import org.apache.flink.table.store.utils.FileIOUtils;
import org.junit.jupiter.api.Test;
@@ -86,7 +86,7 @@ class JavaCodeSplitterTest {
private void runTest(String filename, int maxLength, int maxMembers) {
try {
String code =
- FileUtils.readFileUtf8(
+ FileIOUtils.readFileUtf8(
new File(
JavaCodeSplitterTest.class
.getClassLoader()
@@ -94,7 +94,7 @@ class JavaCodeSplitterTest {
"codesplit/splitter/code/" + filename + ".java")
.toURI()));
String expected =
- FileUtils.readFileUtf8(
+ FileIOUtils.readFileUtf8(
new File(
JavaCodeSplitterTest.class
.getClassLoader()
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/DataFormatTestUtil.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/DataFormatTestUtil.java
index 6e0d0773..48b18441 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/DataFormatTestUtil.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/DataFormatTestUtil.java
@@ -19,8 +19,7 @@ package org.apache.flink.table.store.data;
import org.apache.flink.table.store.memory.MemorySegment;
import org.apache.flink.table.store.types.RowType;
-
-import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.flink.table.store.utils.StringUtils;
import static org.assertj.core.api.Assertions.assertThat;
@@ -55,8 +54,8 @@ public class DataFormatTestUtil {
// header (8 bytes) + 2 * string in fixed-length part (8 bytes each)
BinaryRow row = new BinaryRow(2);
BinaryRowWriter writer = new BinaryRowWriter(row);
- writer.writeString(0, BinaryString.fromString(RandomStringUtils.randomNumeric(2)));
- writer.writeString(1, BinaryString.fromString(RandomStringUtils.randomNumeric(2)));
+ writer.writeString(0, BinaryString.fromString(StringUtils.randomNumericString(2)));
+ writer.writeString(1, BinaryString.fromString(StringUtils.randomNumericString(2)));
writer.complete();
return row;
}
@@ -68,8 +67,8 @@ public class DataFormatTestUtil {
// 64 byte length string (8 bytes in fixed-length, 64 bytes in variable-length)
BinaryRow row = new BinaryRow(2);
BinaryRowWriter writer = new BinaryRowWriter(row);
- writer.writeString(0, BinaryString.fromString(RandomStringUtils.randomNumeric(72)));
- writer.writeString(1, BinaryString.fromString(RandomStringUtils.randomNumeric(64)));
+ writer.writeString(0, BinaryString.fromString(StringUtils.randomNumericString(72)));
+ writer.writeString(1, BinaryString.fromString(StringUtils.randomNumericString(64)));
writer.complete();
return row;
}
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/SerializerTestBase.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/SerializerTestBase.java
index 89d62de9..3ee86c5a 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/SerializerTestBase.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/SerializerTestBase.java
@@ -37,7 +37,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CyclicBarrier;
-import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.table.store.utils.Preconditions.checkArgument;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/datagen/DataGeneratorMapper.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/datagen/DataGeneratorMapper.java
index 02951323..1f44b97f 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/datagen/DataGeneratorMapper.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/datagen/DataGeneratorMapper.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.datagen;
-import org.apache.flink.util.function.SerializableFunction;
+import org.apache.flink.table.store.utils.SerializableFunction;
/** Utility for mapping the output of a {@link DataGenerator}. */
public class DataGeneratorMapper<A, B> implements DataGenerator<B> {
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/datagen/DataGeneratorMapper.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/VersionedSerializerWrapper.java
similarity index 50%
copy from flink-table-store-common/src/test/java/org/apache/flink/table/store/datagen/DataGeneratorMapper.java
copy to flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/VersionedSerializerWrapper.java
index 02951323..476b5700 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/datagen/DataGeneratorMapper.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/VersionedSerializerWrapper.java
@@ -16,34 +16,34 @@
* limitations under the License.
*/
-package org.apache.flink.table.store.datagen;
+package org.apache.flink.table.store.connector;
-import org.apache.flink.util.function.SerializableFunction;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.table.store.data.serializer.VersionedSerializer;
-/** Utility for mapping the output of a {@link DataGenerator}. */
-public class DataGeneratorMapper<A, B> implements DataGenerator<B> {
+import java.io.IOException;
- private final DataGenerator<A> generator;
+/** Wrap a {@link VersionedSerializer} to {@link SimpleVersionedSerializer}. */
+public class VersionedSerializerWrapper<T> implements SimpleVersionedSerializer<T> {
- private final SerializableFunction<A, B> mapper;
+ private final VersionedSerializer<T> serializer;
- public DataGeneratorMapper(DataGenerator<A> generator, SerializableFunction<A, B> mapper) {
- this.generator = generator;
- this.mapper = mapper;
+ public VersionedSerializerWrapper(VersionedSerializer<T> serializer) {
+ this.serializer = serializer;
}
@Override
- public void open() {
- generator.open();
+ public int getVersion() {
+ return serializer.getVersion();
}
@Override
- public boolean hasNext() {
- return generator.hasNext();
+ public byte[] serialize(T obj) throws IOException {
+ return serializer.serialize(obj);
}
@Override
- public B next() {
- return mapper.apply(generator.next());
+ public T deserialize(int version, byte[] serialized) throws IOException {
+ return serializer.deserialize(version, serialized);
}
}
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/FileStoreSink.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/FileStoreSink.java
index 0151f18d..739f8ffe 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/FileStoreSink.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/FileStoreSink.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.store.connector.sink;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.store.connector.VersionedSerializerWrapper;
import org.apache.flink.table.store.file.manifest.ManifestCommittableSerializer;
import org.apache.flink.table.store.file.operation.Lock;
import org.apache.flink.table.store.table.FileStoreTable;
@@ -73,6 +74,7 @@ public class FileStoreSink extends FlinkSink {
@Override
protected CommittableStateManager createCommittableStateManager() {
- return new RestoreAndFailCommittableStateManager(ManifestCommittableSerializer::new);
+ return new RestoreAndFailCommittableStateManager(
+ () -> new VersionedSerializerWrapper<>(new ManifestCommittableSerializer()));
}
}
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTest.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTest.java
index 52e02297..1fb1de45 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTest.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.store.connector.VersionedSerializerWrapper;
import org.apache.flink.table.store.data.GenericRow;
import org.apache.flink.table.store.file.manifest.ManifestCommittableSerializer;
import org.apache.flink.table.store.file.utils.SnapshotManager;
@@ -199,7 +200,9 @@ public class CommitterOperatorTest extends CommitterOperatorTestBase {
initialCommitUser,
user -> new StoreCommitter(table.newCommit(user)),
new RestoreAndFailCommittableStateManager(
- ManifestCommittableSerializer::new));
+ () ->
+ new VersionedSerializerWrapper<>(
+ new ManifestCommittableSerializer())));
return createTestHarness(operator);
}
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTestBase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTestBase.java
index c01a408d..5dde7d0f 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTestBase.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTestBase.java
@@ -33,7 +33,7 @@ import org.apache.flink.table.store.table.source.TableRead;
import org.apache.flink.table.store.types.DataType;
import org.apache.flink.table.store.types.DataTypes;
import org.apache.flink.table.store.types.RowType;
-import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.table.store.utils.CloseableIterator;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.io.TempDir;
diff --git a/flink-table-store-core/pom.xml b/flink-table-store-core/pom.xml
index 3b54e349..db5f95ca 100644
--- a/flink-table-store-core/pom.xml
+++ b/flink-table-store-core/pom.xml
@@ -46,13 +46,6 @@ under the License.
<!-- flink dependencies -->
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-jackson</artifactId>
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/append/AppendOnlyWriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/append/AppendOnlyWriter.java
index eac97aa7..e92caaa4 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/append/AppendOnlyWriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/append/AppendOnlyWriter.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.store.file.append;
-import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.file.compact.CompactManager;
import org.apache.flink.table.store.file.io.CompactIncrement;
@@ -32,6 +31,7 @@ import org.apache.flink.table.store.format.FileFormat;
import org.apache.flink.table.store.fs.FileIO;
import org.apache.flink.table.store.types.RowKind;
import org.apache.flink.table.store.types.RowType;
+import org.apache.flink.table.store.utils.LongCounter;
import org.apache.flink.table.store.utils.Preconditions;
import java.util.ArrayList;
@@ -122,7 +122,7 @@ public class AppendOnlyWriter implements RecordWriter<InternalRow> {
flushedFiles.addAll(writer.result());
// Reopen the writer to accept further records.
- seqNumCounter.resetLocal();
+ seqNumCounter.reset();
seqNumCounter.add(getMaxSequenceNumber(flushedFiles) + 1);
writer = createRollingRowWriter();
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/AbstractCatalog.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/AbstractCatalog.java
index 8af5ad89..a0a8be93 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/AbstractCatalog.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/AbstractCatalog.java
@@ -24,8 +24,7 @@ import org.apache.flink.table.store.fs.Path;
import org.apache.flink.table.store.table.FileStoreTableFactory;
import org.apache.flink.table.store.table.Table;
import org.apache.flink.table.store.table.system.SystemTableLoader;
-
-import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.table.store.utils.StringUtils;
/** Common implementation of {@link Catalog}. */
public abstract class AbstractCatalog implements Catalog {
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/FileWriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/FileWriter.java
index d8feb947..09677c01 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/FileWriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/FileWriter.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.store.file.io;
-import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.table.store.utils.CloseableIterator;
import java.io.Closeable;
import java.io.IOException;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/RowDataFileWriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/RowDataFileWriter.java
index 8987745d..c9281e84 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/RowDataFileWriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/RowDataFileWriter.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.store.file.io;
-import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.file.stats.BinaryTableStats;
import org.apache.flink.table.store.file.stats.FieldStatsArraySerializer;
@@ -28,6 +27,7 @@ import org.apache.flink.table.store.format.FormatWriterFactory;
import org.apache.flink.table.store.fs.FileIO;
import org.apache.flink.table.store.fs.Path;
import org.apache.flink.table.store.types.RowType;
+import org.apache.flink.table.store.utils.LongCounter;
import javax.annotation.Nullable;
@@ -72,8 +72,8 @@ public class RowDataFileWriter extends StatsCollectingSingleFileWriter<InternalR
fileIO.getFileSize(path),
recordCount(),
stats,
- seqNumCounter.getLocalValue() - super.recordCount(),
- seqNumCounter.getLocalValue() - 1,
+ seqNumCounter.getValue() - super.recordCount(),
+ seqNumCounter.getValue() - 1,
schemaId);
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/RowDataRollingFileWriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/RowDataRollingFileWriter.java
index 980ba068..e8d6325f 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/RowDataRollingFileWriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/RowDataRollingFileWriter.java
@@ -19,11 +19,11 @@
package org.apache.flink.table.store.file.io;
-import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.format.FileFormat;
import org.apache.flink.table.store.fs.FileIO;
import org.apache.flink.table.store.types.RowType;
+import org.apache.flink.table.store.utils.LongCounter;
/** {@link RollingFileWriter} for data files containing {@link InternalRow}. */
public class RowDataRollingFileWriter extends RollingFileWriter<InternalRow, DataFileMeta> {
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestCommittableSerializer.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestCommittableSerializer.java
index 09e8fd4b..4a3be4ab 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestCommittableSerializer.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestCommittableSerializer.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.file.manifest;
-import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.table.store.data.serializer.VersionedSerializer;
import org.apache.flink.table.store.io.DataInputDeserializer;
import org.apache.flink.table.store.io.DataOutputViewStreamWrapper;
import org.apache.flink.table.store.table.sink.FileCommittable;
@@ -30,9 +30,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-/** {@link SimpleVersionedSerializer} for {@link ManifestCommittable}. */
-public class ManifestCommittableSerializer
- implements SimpleVersionedSerializer<ManifestCommittable> {
+/** {@link VersionedSerializer} for {@link ManifestCommittable}. */
+public class ManifestCommittableSerializer implements VersionedSerializer<ManifestCommittable> {
private static final int CURRENT_VERSION = 2;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestList.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestList.java
index 93432b3a..8efa7cb4 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestList.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestList.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.file.manifest;
-import org.apache.flink.api.common.serialization.BulkWriter;
import org.apache.flink.table.store.file.utils.FileStorePathFactory;
import org.apache.flink.table.store.file.utils.FileUtils;
import org.apache.flink.table.store.file.utils.VersionedObjectSerializer;
@@ -104,10 +103,7 @@ public class ManifestList {
fileIO.deleteQuietly(pathFactory.toManifestListPath(fileName));
}
- /**
- * Creator of {@link ManifestList}. It reueses {@link FormatReaderFactory} and {@link
- * BulkWriter.Factory} from {@link FileFormat}.
- */
+ /** Creator of {@link ManifestList}. */
public static class Factory {
private final FileIO fileIO;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AppendOnlyFileStoreWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AppendOnlyFileStoreWrite.java
index fcb8e3ab..1723487d 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AppendOnlyFileStoreWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AppendOnlyFileStoreWrite.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.file.operation;
-import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.data.BinaryRow;
import org.apache.flink.table.store.data.InternalRow;
@@ -37,6 +36,7 @@ import org.apache.flink.table.store.format.FileFormat;
import org.apache.flink.table.store.fs.FileIO;
import org.apache.flink.table.store.table.source.DataSplit;
import org.apache.flink.table.store.types.RowType;
+import org.apache.flink.table.store.utils.LongCounter;
import java.util.Collections;
import java.util.LinkedList;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/sort/BinaryExternalSortBuffer.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/sort/BinaryExternalSortBuffer.java
index a635998d..8cb08a76 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/sort/BinaryExternalSortBuffer.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/sort/BinaryExternalSortBuffer.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.file.sort;
-import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.store.annotation.VisibleForTesting;
import org.apache.flink.table.store.codegen.RecordComparator;
import org.apache.flink.table.store.data.BinaryRow;
import org.apache.flink.table.store.data.InternalRow;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FatalExitExceptionHandler.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FatalExitExceptionHandler.java
index 44e4707d..0fc70f69 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FatalExitExceptionHandler.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FatalExitExceptionHandler.java
@@ -18,8 +18,6 @@
package org.apache.flink.table.store.file.utils;
-import org.apache.flink.core.security.FlinkSecurityManager;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,6 +34,7 @@ public final class FatalExitExceptionHandler implements Thread.UncaughtException
public static final FatalExitExceptionHandler INSTANCE = new FatalExitExceptionHandler();
public static final int EXIT_CODE = -17;
+ @SuppressWarnings("finally")
@Override
public void uncaughtException(Thread t, Throwable e) {
try {
@@ -45,7 +44,7 @@ public final class FatalExitExceptionHandler implements Thread.UncaughtException
e);
ThreadUtils.errorLogThreadDump(LOG);
} finally {
- FlinkSecurityManager.forceProcessExit(EXIT_CODE);
+ System.exit(EXIT_CODE);
}
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FileStorePathFactory.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FileStorePathFactory.java
index 34c87409..ebaeae4a 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FileStorePathFactory.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FileStorePathFactory.java
@@ -18,8 +18,8 @@
package org.apache.flink.table.store.file.utils;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.table.store.CoreOptions;
+import org.apache.flink.table.store.annotation.VisibleForTesting;
import org.apache.flink.table.store.data.BinaryRow;
import org.apache.flink.table.store.file.io.DataFilePathFactory;
import org.apache.flink.table.store.fs.Path;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/PartitionPathUtils.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/PartitionPathUtils.java
index 49f7e333..0e774877 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/PartitionPathUtils.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/PartitionPathUtils.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.file.utils;
-import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.store.fs.Path;
import java.util.BitSet;
import java.util.LinkedHashMap;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
index 425871dd..e76cdee3 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
@@ -18,8 +18,8 @@
package org.apache.flink.table.store.table;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.table.store.CoreOptions;
+import org.apache.flink.table.store.annotation.VisibleForTesting;
import org.apache.flink.table.store.file.FileStore;
import org.apache.flink.table.store.file.schema.SchemaManager;
import org.apache.flink.table.store.file.schema.TableSchema;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/FileCommittableSerializer.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/FileCommittableSerializer.java
index 60132f08..5c4bcfb3 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/FileCommittableSerializer.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/FileCommittableSerializer.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.table.sink;
-import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.table.store.data.serializer.VersionedSerializer;
import org.apache.flink.table.store.file.io.CompactIncrement;
import org.apache.flink.table.store.file.io.DataFileMetaSerializer;
import org.apache.flink.table.store.file.io.NewFilesIncrement;
@@ -35,8 +35,8 @@ import java.util.List;
import static org.apache.flink.table.store.file.utils.SerializationUtils.deserializeBinaryRow;
import static org.apache.flink.table.store.file.utils.SerializationUtils.serializeBinaryRow;
-/** {@link SimpleVersionedSerializer} for {@link FileCommittable}. */
-public class FileCommittableSerializer implements SimpleVersionedSerializer<FileCommittable> {
+/** {@link VersionedSerializer} for {@link FileCommittable}. */
+public class FileCommittableSerializer implements VersionedSerializer<FileCommittable> {
private static final int CURRENT_VERSION = 2;
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/KeyValueFileReadWriteTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/KeyValueFileReadWriteTest.java
index ecdb1760..90ce678d 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/KeyValueFileReadWriteTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/KeyValueFileReadWriteTest.java
@@ -41,7 +41,7 @@ import org.apache.flink.table.store.types.DataType;
import org.apache.flink.table.store.types.IntType;
import org.apache.flink.table.store.types.RowType;
import org.apache.flink.table.store.types.VarCharType;
-import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.table.store.utils.CloseableIterator;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/RollingFileWriterTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/RollingFileWriterTest.java
index 0323ca90..9c4529af 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/RollingFileWriterTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/RollingFileWriterTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.file.io;
-import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.data.GenericRow;
import org.apache.flink.table.store.data.InternalRow;
@@ -29,6 +28,7 @@ import org.apache.flink.table.store.options.Options;
import org.apache.flink.table.store.types.DataType;
import org.apache.flink.table.store.types.IntType;
import org.apache.flink.table.store.types.RowType;
+import org.apache.flink.table.store.utils.LongCounter;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
diff --git a/flink-table-store-format/pom.xml b/flink-table-store-format/pom.xml
index abab49c9..da5f0a89 100644
--- a/flink-table-store-format/pom.xml
+++ b/flink-table-store-format/pom.xml
@@ -44,13 +44,6 @@ under the License.
<scope>provided</scope>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
-
<!-- Hadoop -->
<dependency>
diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/avro/AvroBulkWriter.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/avro/AvroBulkWriter.java
index d1240e87..9e4b73c6 100644
--- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/avro/AvroBulkWriter.java
+++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/avro/AvroBulkWriter.java
@@ -18,13 +18,11 @@
package org.apache.flink.table.store.format.avro;
-import org.apache.flink.api.common.serialization.BulkWriter;
-
import org.apache.avro.file.DataFileWriter;
import java.io.IOException;
-/** A simple {@link BulkWriter} implementation that wraps an Avro {@link DataFileWriter}. */
+/** A simple writer implementation that wraps an Avro {@link DataFileWriter}. */
public class AvroBulkWriter<T> {
/** The underlying Avro writer. */
diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
index 18a4c827..21ee6777 100644
--- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
+++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
@@ -18,8 +18,7 @@
package org.apache.flink.table.store.format.orc;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.serialization.BulkWriter;
+import org.apache.flink.table.store.annotation.VisibleForTesting;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.format.FileFormat;
@@ -110,8 +109,8 @@ public class OrcFileFormat extends FileFormat {
* <p>TODO: The {@link ThreadLocalClassLoaderConfiguration} in {@link OrcWriterFactory} should
* be removed after https://issues.apache.org/jira/browse/ORC-653 is fixed.
*
- * @param type The data type for the {@link BulkWriter}
- * @return The factory of the {@link BulkWriter}
+ * @param type The data type for the writer
+ * @return The factory of the writer
*/
@Override
public FormatWriterFactory createWriterFactory(RowType type) {
diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcReaderFactory.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcReaderFactory.java
index c4463f20..8e531721 100644
--- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcReaderFactory.java
+++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcReaderFactory.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.format.orc;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.data.columnar.ColumnVector;
import org.apache.flink.table.store.data.columnar.ColumnarRow;
@@ -32,6 +31,7 @@ import org.apache.flink.table.store.fs.FileIO;
import org.apache.flink.table.store.fs.Path;
import org.apache.flink.table.store.types.DataType;
import org.apache.flink.table.store.types.RowType;
+import org.apache.flink.table.store.utils.Pair;
import org.apache.flink.table.store.utils.Pool;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
@@ -254,14 +254,14 @@ public class OrcReaderFactory implements FormatReaderFactory {
org.apache.orc.Reader orcReader = createReader(conf, fileIO, path);
// get offset and length for the stripes that start in the split
- Tuple2<Long, Long> offsetAndLength =
+ Pair<Long, Long> offsetAndLength =
getOffsetAndLengthForSplit(splitStart, splitLength, orcReader.getStripes());
// create ORC row reader configuration
org.apache.orc.Reader.Options options =
new org.apache.orc.Reader.Options()
.schema(schema)
- .range(offsetAndLength.f0, offsetAndLength.f1)
+ .range(offsetAndLength.getLeft(), offsetAndLength.getRight())
.useZeroCopy(OrcConf.USE_ZEROCOPY.getBoolean(conf))
.skipCorruptRecords(OrcConf.SKIP_CORRUPT_DATA.getBoolean(conf))
.tolerateMissingSchema(OrcConf.TOLERATE_MISSING_SCHEMA.getBoolean(conf));
@@ -298,7 +298,7 @@ public class OrcReaderFactory implements FormatReaderFactory {
return reader.nextBatch(rowBatch);
}
- private static Tuple2<Long, Long> getOffsetAndLengthForSplit(
+ private static Pair<Long, Long> getOffsetAndLengthForSplit(
long splitStart, long splitLength, List<StripeInformation> stripes) {
long splitEnd = splitStart + splitLength;
long readStart = Long.MAX_VALUE;
@@ -314,9 +314,9 @@ public class OrcReaderFactory implements FormatReaderFactory {
if (readStart < Long.MAX_VALUE) {
// at least one split is included
- return Tuple2.of(readStart, readEnd - readStart);
+ return Pair.of(readStart, readEnd - readStart);
} else {
- return Tuple2.of(0L, 0L);
+ return Pair.of(0L, 0L);
}
}
diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcWriterFactory.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcWriterFactory.java
index de14e7db..37b58b6a 100644
--- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcWriterFactory.java
+++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcWriterFactory.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.format.orc;
-import org.apache.flink.api.common.serialization.BulkWriter;
import org.apache.flink.table.store.annotation.VisibleForTesting;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.format.FormatWriter;
@@ -42,7 +41,7 @@ import java.util.UUID;
import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
/**
- * A factory that creates an ORC {@link BulkWriter}. The factory takes a user supplied {@link
+ * A factory that creates an ORC {@link FormatWriter}. The factory takes a user supplied {@link
* Vectorizer} implementation to convert the element into an {@link
* org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch}.
*/
diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/SerializableHadoopConfigWrapper.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/SerializableHadoopConfigWrapper.java
index 409f7c59..82c061d4 100644
--- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/SerializableHadoopConfigWrapper.java
+++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/SerializableHadoopConfigWrapper.java
@@ -18,8 +18,8 @@
package org.apache.flink.table.store.format.orc;
-import org.apache.flink.core.memory.DataInputDeserializer;
-import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.table.store.io.DataInputDeserializer;
+import org.apache.flink.table.store.io.DataOutputSerializer;
import org.apache.hadoop.conf.Configuration;
diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/writer/OrcBulkWriter.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/writer/OrcBulkWriter.java
index 63b7f79c..1ded35ad 100644
--- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/writer/OrcBulkWriter.java
+++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/writer/OrcBulkWriter.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.format.orc.writer;
-import org.apache.flink.api.common.serialization.BulkWriter;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.format.FormatWriter;
@@ -29,7 +28,7 @@ import java.io.IOException;
import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
-/** A {@link BulkWriter} implementation that writes data in ORC format. */
+/** A {@link FormatWriter} implementation that writes data in ORC format. */
public class OrcBulkWriter implements FormatWriter {
private final Writer writer;
diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/parquet/ParquetFileFormat.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/parquet/ParquetFileFormat.java
index 864dfd19..48ee1412 100644
--- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/parquet/ParquetFileFormat.java
+++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/parquet/ParquetFileFormat.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.format.parquet;
-import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.store.annotation.VisibleForTesting;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.format.FileFormat;
import org.apache.flink.table.store.format.FileStatsExtractor;
@@ -53,8 +53,7 @@ public class ParquetFileFormat extends FileFormat {
public FormatReaderFactory createReaderFactory(
RowType type, int[][] projection, List<Predicate> filters) {
return new ParquetReaderFactory(
- getParquetConfiguration(formatOptions),
- (RowType) Projection.of(projection).project(type));
+ getParquetConfiguration(formatOptions), Projection.of(projection).project(type));
}
@Override
diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/parquet/writer/ParquetBulkWriter.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/parquet/writer/ParquetBulkWriter.java
index 9dbf2114..a19baf89 100644
--- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/parquet/writer/ParquetBulkWriter.java
+++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/parquet/writer/ParquetBulkWriter.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.format.parquet.writer;
-import org.apache.flink.api.common.serialization.BulkWriter;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.format.FormatWriter;
@@ -28,7 +27,7 @@ import java.io.IOException;
import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
-/** A simple {@link BulkWriter} implementation that wraps a {@link ParquetWriter}. */
+/** A simple {@link FormatWriter} implementation that wraps a {@link ParquetWriter}. */
public class ParquetBulkWriter implements FormatWriter {
/** The ParquetWriter to write to. */
diff --git a/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/parquet/ParquetReadWriteTest.java b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/parquet/ParquetReadWriteTest.java
index b619a712..f7d48345 100644
--- a/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/parquet/ParquetReadWriteTest.java
+++ b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/parquet/ParquetReadWriteTest.java
@@ -74,8 +74,6 @@ import static org.assertj.core.api.Assertions.assertThat;
public class ParquetReadWriteTest {
private static final LocalDateTime BASE_TIME = LocalDateTime.now();
- private static final org.apache.flink.configuration.Configuration EMPTY_CONF =
- new org.apache.flink.configuration.Configuration();
private static final RowType ROW_TYPE =
RowType.builder()
diff --git a/flink-table-store-hive/flink-table-store-hive-connector/pom.xml b/flink-table-store-hive/flink-table-store-hive-connector/pom.xml
index 0050b6a4..3b34df0b 100644
--- a/flink-table-store-hive/flink-table-store-hive-connector/pom.xml
+++ b/flink-table-store-hive/flink-table-store-hive-connector/pom.xml
@@ -148,6 +148,20 @@ under the License.
<!-- dependencies for IT cases -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-common</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
diff --git a/flink-table-store-hive/flink-table-store-hive-connector/src/main/java/org/apache/flink/table/store/TableStoreJobConf.java b/flink-table-store-hive/flink-table-store-hive-connector/src/main/java/org/apache/flink/table/store/TableStoreJobConf.java
index cf69b566..a7bc5298 100644
--- a/flink-table-store-hive/flink-table-store-hive-connector/src/main/java/org/apache/flink/table/store/TableStoreJobConf.java
+++ b/flink-table-store-hive/flink-table-store-hive-connector/src/main/java/org/apache/flink/table/store/TableStoreJobConf.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.store;
import org.apache.flink.table.store.file.utils.JsonSerdeUtil;
+import org.apache.flink.table.store.options.Options;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
@@ -59,15 +60,8 @@ public class TableStoreJobConf {
return jobConf.get(INTERNAL_LOCATION);
}
- @SuppressWarnings("unchecked")
- public org.apache.flink.configuration.Configuration getCatalogConfig() {
- return org.apache.flink.configuration.Configuration.fromMap(
- JsonSerdeUtil.fromJson(jobConf.get(INTERNAL_CATALOG_CONFIG), Map.class));
- }
-
/** Extract table store catalog conf from Hive conf. */
- public static org.apache.flink.configuration.Configuration extractCatalogConfig(
- Configuration hiveConf) {
+ public static Options extractCatalogConfig(Configuration hiveConf) {
Map<String, String> configMap = new HashMap<>();
for (Map.Entry<String, String> entry : hiveConf) {
String name = entry.getKey();
@@ -77,6 +71,6 @@ public class TableStoreJobConf {
configMap.put(name, value);
}
}
- return org.apache.flink.configuration.Configuration.fromMap(configMap);
+ return Options.fromMap(configMap);
}
}
diff --git a/flink-table-store-hive/flink-table-store-hive-connector/src/main/java/org/apache/flink/table/store/hive/HiveSchema.java b/flink-table-store-hive/flink-table-store-hive-connector/src/main/java/org/apache/flink/table/store/hive/HiveSchema.java
index 7c9871a8..0abade0e 100644
--- a/flink-table-store-hive/flink-table-store-hive-connector/src/main/java/org/apache/flink/table/store/hive/HiveSchema.java
+++ b/flink-table-store-hive/flink-table-store-hive-connector/src/main/java/org/apache/flink/table/store/hive/HiveSchema.java
@@ -18,9 +18,9 @@
package org.apache.flink.table.store.hive;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.file.schema.TableSchema;
+import org.apache.flink.table.store.fs.Path;
import org.apache.flink.table.store.options.CatalogOptions;
import org.apache.flink.table.store.options.Options;
import org.apache.flink.table.store.table.FileStoreTableFactory;
diff --git a/flink-table-store-shade/pom.xml b/flink-table-store-shade/pom.xml
index 740a7f1c..6352f6c7 100644
--- a/flink-table-store-shade/pom.xml
+++ b/flink-table-store-shade/pom.xml
@@ -47,70 +47,6 @@ under the License.
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${flink.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-metrics-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.esotericsoftware.kryo</groupId>
- <artifactId>kryo</artifactId>
- </exclusion>
- <exclusion>
- <groupId>commons-collections</groupId>
- <artifactId>commons-collections</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-compress</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-common</artifactId>
- <version>${flink.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-api-java</artifactId>
- <version>${flink.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>${flink.table.runtime}</artifactId>
- <version>${flink.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
<!-- codegen runtime dependencies -->
<dependency>
@@ -119,32 +55,6 @@ under the License.
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>${flink.streaming.java}</artifactId>
- <version>${flink.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <!-- filesystem runtime dependencies -->
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-hadoop-fs</artifactId>
- <version>${flink.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
<!-- format runtime dependencies -->
<dependency>
@@ -153,32 +63,6 @@ under the License.
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-files</artifactId>
- <version>${flink.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <!-- "runtime" runtime dependencies, mainly for quick sorting memory buffer -->
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime</artifactId>
- <version>${flink.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
<dependency>
<groupId>org.xerial.snappy</groupId>
<artifactId>snappy-java</artifactId>
@@ -202,7 +86,7 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-guava</artifactId>
- <version>30.1.1-jre-${flink.shaded.version}</version>
+ <version>${flink.shaded.guava.version}-${flink.shaded.version}</version>
</dependency>
<dependency>
@@ -228,19 +112,10 @@ under the License.
<artifactSet>
<includes combine.children="append">
<!-- Flink -->
- <include>org.apache.flink:flink-core</include>
- <include>org.apache.flink:flink-runtime</include>
- <include>org.apache.flink:${flink.streaming.java}</include>
- <include>org.apache.flink:flink-hadoop-fs</include>
<include>org.apache.flink:flink-shaded-jackson</include>
<include>org.apache.flink:flink-shaded-guava</include>
<include>org.apache.flink:flink-shaded-asm-9</include>
- <!-- Flink Table -->
- <include>org.apache.flink:flink-table-common</include>
- <include>org.apache.flink:flink-table-api-java</include>
- <include>org.apache.flink:${flink.table.runtime}</include>
-
<!-- Flink Table Store -->
<include>org.apache.flink:flink-table-store-common</include>
<include>org.apache.flink:flink-table-store-core</include>
diff --git a/pom.xml b/pom.xml
index d65a6144..4c1c6c04 100644
--- a/pom.xml
+++ b/pom.xml
@@ -75,6 +75,7 @@ under the License.
<flink.connector.version>1.15</flink.connector.version>
<flink.shaded.version>15.0</flink.shaded.version>
<flink.shaded.jackson.version>2.12.4</flink.shaded.jackson.version>
+ <flink.shaded.guava.version>30.1.1-jre</flink.shaded.guava.version>
<hadoop.version>2.8.5</hadoop.version>
<scala.version>2.12.7</scala.version>
<scala.binary.version>2.12</scala.binary.version>