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/15 11:42:35 UTC
[flink-table-store] branch master updated: [FLINK-31078] flink-table-planner free for flink-table-store-core
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 de4395fb [FLINK-31078] flink-table-planner free for flink-table-store-core
de4395fb is described below
commit de4395fb192b21216f29460abea48cbbe72f5d47
Author: Shammon FY <zj...@gmail.com>
AuthorDate: Wed Feb 15 19:42:29 2023 +0800
[FLINK-31078] flink-table-planner free for flink-table-store-core
This closes #539
---
flink-table-store-common/pom.xml | 22 +++++++-
.../flink/table/store/utils/StringUtils.java | 66 ++++++++++++++++++++++
flink-table-store-core/pom.xml | 7 ---
.../store/file/append/AppendOnlyWriterTest.java | 16 +++---
.../table/store/file/utils/FailingFileIO.java | 5 +-
.../table/store/file/utils/TraceableFileIO.java | 35 ++++++++----
.../table/ChangelogWithKeyFileStoreTableTest.java | 23 ++++----
.../store/table/SchemaEvolutionTableTestBase.java | 2 +-
8 files changed, 134 insertions(+), 42 deletions(-)
diff --git a/flink-table-store-common/pom.xml b/flink-table-store-common/pom.xml
index 45c3ed4d..5f7ed041 100644
--- a/flink-table-store-common/pom.xml
+++ b/flink-table-store-common/pom.xml
@@ -53,14 +53,12 @@ under the License.
<groupId>org.codehaus.janino</groupId>
<artifactId>janino</artifactId>
<version>${janino.version}</version>
- <scope>provided</scope>
</dependency>
<dependency>
<groupId>org.codehaus.janino</groupId>
<artifactId>commons-compiler</artifactId>
<version>${janino.version}</version>
- <scope>provided</scope>
</dependency>
<dependency>
@@ -240,13 +238,33 @@ under the License.
<artifactSet>
<includes combine.children="append">
<include>org.antlr:antlr4-runtime</include>
+ <include>org.codehaus.janino:*</include>
</includes>
</artifactSet>
+ <filters>
+ <filter>
+ <!-- Do not copy the signatures in the META-INF folder and factories in janino.
+ Otherwise, this might cause SecurityExceptions or be incompatible with spark/hive when using the JAR. -->
+ <artifact>org.codehaus.janino:*</artifact>
+ <excludes>
+ <exclude>META-INF/**</exclude>
+ <exclude>org.codehaus.commons.compiler.properties</exclude>
+ </excludes>
+ </filter>
+ </filters>
<relocations>
<relocation>
<pattern>org.antlr.v4.runtime</pattern>
<shadedPattern>org.apache.flink.table.store.shaded.org.antlr.v4.runtime</shadedPattern>
</relocation>
+ <relocation>
+ <pattern>org.codehaus.janino</pattern>
+ <shadedPattern>org.apache.flink.table.store.shaded.org.codehaus.janino</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.codehaus.commons</pattern>
+ <shadedPattern>org.apache.flink.table.store.shaded.org.codehaus.commons</shadedPattern>
+ </relocation>
</relocations>
</configuration>
</execution>
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 47b25e80..37faa38f 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
@@ -23,7 +23,9 @@ import org.apache.flink.table.store.memory.MemorySegmentUtils;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Iterator;
import java.util.List;
+import java.util.Objects;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
@@ -47,6 +49,9 @@ public class StringUtils {
*/
public static final int INDEX_NOT_FOUND = -1;
+ /** The empty String {@code ""}. */
+ public static final String EMPTY = "";
+
/**
* Concatenates input strings together into a single string. Returns NULL if any argument is
* NULL.
@@ -456,4 +461,65 @@ public class StringUtils {
}
return list.toArray(new String[list.size()]);
}
+
+ /**
+ * Joins the elements of the provided {@code Iterable} into a single String containing the
+ * provided elements.
+ *
+ * <p>No delimiter is added before or after the list. A {@code null} separator is the same as an
+ * empty String ("").
+ *
+ * @param iterable the {@code Iterable} providing the values to join together, may be null
+ * @param separator the separator character to use, null treated as ""
+ * @return the joined String, {@code null} if null iterator input
+ */
+ public static String join(final Iterable<?> iterable, final String separator) {
+ if (iterable == null) {
+ return null;
+ }
+ return join(iterable.iterator(), separator);
+ }
+
+ /**
+ * Joins the elements of the provided {@code Iterator} into a single String containing the
+ * provided elements.
+ *
+ * <p>No delimiter is added before or after the list. A {@code null} separator is the same as an
+ * empty String ("").
+ *
+ * @param iterator the {@code Iterator} of values to join together, may be null
+ * @param separator the separator character to use, null treated as ""
+ * @return the joined String, {@code null} if null iterator input
+ */
+ public static String join(final Iterator<?> iterator, final String separator) {
+
+ // handle null, zero and one elements before building a buffer
+ if (iterator == null) {
+ return null;
+ }
+ if (!iterator.hasNext()) {
+ return EMPTY;
+ }
+ final Object first = iterator.next();
+ if (!iterator.hasNext()) {
+ return Objects.toString(first);
+ }
+
+ // two or more elements
+ final StringBuilder buf = new StringBuilder(256); // Java default is 16, probably too small
+ if (first != null) {
+ buf.append(first);
+ }
+
+ while (iterator.hasNext()) {
+ if (separator != null) {
+ buf.append(separator);
+ }
+ final Object obj = iterator.next();
+ if (obj != null) {
+ buf.append(obj);
+ }
+ }
+ return buf.toString();
+ }
}
diff --git a/flink-table-store-core/pom.xml b/flink-table-store-core/pom.xml
index db5f95ca..4c534972 100644
--- a/flink-table-store-core/pom.xml
+++ b/flink-table-store-core/pom.xml
@@ -68,13 +68,6 @@ under the License.
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
- <version>${flink.version}</version>
- <scope>test</scope>
- </dependency>
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-store-format</artifactId>
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/append/AppendOnlyWriterTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/append/AppendOnlyWriterTest.java
index cff50609..0c06c00b 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/append/AppendOnlyWriterTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/append/AppendOnlyWriterTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.store.file.append;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.data.BinaryRow;
import org.apache.flink.table.store.data.BinaryString;
@@ -39,6 +38,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.table.store.utils.Pair;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -127,7 +127,7 @@ public class AppendOnlyWriterTest {
@Test
public void testMultipleCommits() throws Exception {
RecordWriter<InternalRow> writer =
- createWriter(1024 * 1024L, true, Collections.emptyList()).f0;
+ createWriter(1024 * 1024L, true, Collections.emptyList()).getLeft();
// Commit 5 continues txn.
for (int txn = 0; txn < 5; txn += 1) {
@@ -234,10 +234,10 @@ public class AppendOnlyWriterTest {
// increase target file size to test compaction
long targetFileSize = 1024 * 1024L;
- Tuple2<AppendOnlyWriter, LinkedList<DataFileMeta>> writerAndToCompact =
+ Pair<AppendOnlyWriter, LinkedList<DataFileMeta>> writerAndToCompact =
createWriter(targetFileSize, true, firstInc.newFilesIncrement().newFiles());
- writer = writerAndToCompact.f0;
- LinkedList<DataFileMeta> toCompact = writerAndToCompact.f1;
+ writer = writerAndToCompact.getLeft();
+ LinkedList<DataFileMeta> toCompact = writerAndToCompact.getRight();
assertThat(toCompact).containsExactlyElementsOf(firstInc.newFilesIncrement().newFiles());
writer.write(row(id, String.format("%03d", id), PART));
writer.sync();
@@ -298,14 +298,14 @@ public class AppendOnlyWriterTest {
}
private AppendOnlyWriter createEmptyWriter(long targetFileSize) {
- return createWriter(targetFileSize, false, Collections.emptyList()).f0;
+ return createWriter(targetFileSize, false, Collections.emptyList()).getLeft();
}
- private Tuple2<AppendOnlyWriter, LinkedList<DataFileMeta>> createWriter(
+ private Pair<AppendOnlyWriter, LinkedList<DataFileMeta>> createWriter(
long targetFileSize, boolean forceCompact, List<DataFileMeta> scannedFiles) {
FileFormat fileFormat = FileFormat.fromIdentifier(AVRO, new Options());
LinkedList<DataFileMeta> toCompact = new LinkedList<>(scannedFiles);
- return new Tuple2<>(
+ return Pair.of(
new AppendOnlyWriter(
LocalFileIO.create(),
SCHEMA_ID,
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/FailingFileIO.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/FailingFileIO.java
index e2fc24d8..5b6a8d74 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/FailingFileIO.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/FailingFileIO.java
@@ -27,8 +27,7 @@ import org.apache.flink.table.store.fs.PositionOutputStreamWrapper;
import org.apache.flink.table.store.fs.SeekableInputStream;
import org.apache.flink.table.store.fs.SeekableInputStreamWrapper;
import org.apache.flink.table.store.fs.local.LocalFileIO;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.function.RunnableWithException;
+import org.apache.flink.table.store.utils.ExceptionUtils;
import java.io.File;
import java.io.FileNotFoundException;
@@ -187,7 +186,7 @@ public class FailingFileIO extends TraceableFileIO {
}
}
- public static void retryArtificialException(RunnableWithException runnable) throws Exception {
+ public static void retryArtificialException(Runnable runnable) throws Exception {
while (true) {
try {
runnable.run();
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/TraceableFileIO.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/TraceableFileIO.java
index 033e2afb..391645d2 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/TraceableFileIO.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/TraceableFileIO.java
@@ -29,7 +29,6 @@ import org.apache.flink.table.store.fs.SeekableInputStreamWrapper;
import org.apache.flink.table.store.fs.local.LocalFileIO;
import org.apache.flink.table.store.options.CatalogOptions;
import org.apache.flink.table.store.utils.ThreadUtils;
-import org.apache.flink.util.function.SupplierWithException;
import javax.annotation.concurrent.GuardedBy;
@@ -39,6 +38,7 @@ import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
@@ -63,7 +63,15 @@ public class TraceableFileIO implements FileIO {
@Override
public PositionOutputStream newOutputStream(Path f, boolean overwrite) throws IOException {
- return createOutputStream(f, () -> originalFs.newOutputStream(f, overwrite));
+ return createOutputStream(
+ f,
+ () -> {
+ try {
+ return originalFs.newOutputStream(f, overwrite);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
}
@Override
@@ -78,24 +86,30 @@ public class TraceableFileIO implements FileIO {
@Override
public SeekableInputStream newInputStream(Path f) throws IOException {
- return createInputStream(f, () -> originalFs.newInputStream(f));
+ return createInputStream(
+ f,
+ () -> {
+ try {
+ return originalFs.newInputStream(f);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
}
private PositionOutputStream createOutputStream(
- Path f, SupplierWithException<PositionOutputStream, IOException> streamOpener)
- throws IOException {
+ Path f, Supplier<PositionOutputStream> streamOpener) throws IOException {
- final SupplierWithException<OutStream, IOException> wrappedStreamOpener =
+ final Supplier<OutStream> wrappedStreamOpener =
() -> new OutStream(ThreadUtils.currentStackString(), f, streamOpener.get(), this);
return createStream(wrappedStreamOpener, OPEN_OUTPUT_STREAMS);
}
private SeekableInputStream createInputStream(
- Path f, SupplierWithException<SeekableInputStream, IOException> streamOpener)
- throws IOException {
+ Path f, Supplier<SeekableInputStream> streamOpener) throws IOException {
- final SupplierWithException<InStream, IOException> wrappedStreamOpener =
+ final Supplier<InStream> wrappedStreamOpener =
() -> new InStream(ThreadUtils.currentStackString(), f, streamOpener.get(), this);
return createStream(wrappedStreamOpener, OPEN_INPUT_STREAMS);
@@ -131,8 +145,7 @@ public class TraceableFileIO implements FileIO {
return originalFs.exists(f);
}
- private <T> T createStream(
- final SupplierWithException<T, IOException> streamOpener, final HashSet<T> openStreams)
+ private <T> T createStream(final Supplier<T> streamOpener, final HashSet<T> openStreams)
throws IOException {
// open the stream outside the lock.
final T out = streamOpener.get();
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
index 16a65604..f13b7f02 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
@@ -49,7 +49,6 @@ import org.apache.flink.table.store.types.DataTypes;
import org.apache.flink.table.store.types.RowKind;
import org.apache.flink.table.store.types.RowType;
import org.apache.flink.table.store.utils.CompatibilityTestUtils;
-import org.apache.flink.util.function.FunctionWithException;
import org.junit.jupiter.api.Test;
@@ -387,7 +386,7 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
new InputChangelogFollowUpScanner(),
1L);
- FunctionWithException<Integer, Void, Exception> assertNextSnapshot =
+ Function<Integer, Void> assertNextSnapshot =
i -> {
TableScan.Plan plan = enumerator.enumerate();
assertThat(plan).isNotNull();
@@ -395,14 +394,18 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
List<Split> splits = plan.splits();
TableRead read = table.newRead();
for (int j = 0; j < 2; j++) {
- assertThat(
- getResult(
- read,
- splits,
- binaryRow(j + 1),
- 0,
- COMPATIBILITY_CHANGELOG_ROW_TO_STRING))
- .isEqualTo(expected.get(i).get(j));
+ try {
+ assertThat(
+ getResult(
+ read,
+ splits,
+ binaryRow(j + 1),
+ 0,
+ COMPATIBILITY_CHANGELOG_ROW_TO_STRING))
+ .isEqualTo(expected.get(i).get(j));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
return null;
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTableTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTableTestBase.java
index ec59c3fe..22e99b8c 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTableTestBase.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTableTestBase.java
@@ -55,8 +55,8 @@ import org.apache.flink.table.store.types.SmallIntType;
import org.apache.flink.table.store.types.TimestampType;
import org.apache.flink.table.store.types.VarBinaryType;
import org.apache.flink.table.store.types.VarCharType;
+import org.apache.flink.table.store.utils.StringUtils;
-import org.apache.commons.lang3.StringUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.io.TempDir;