You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2020/11/03 15:34:37 UTC
[flink] 03/10: [FLINK-19802][connector files] Let BulkFormat
createReader and restoreReader methods accept Splits directly
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a2849b327635233d34686a9ab95cf3486ac08ae5
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Sun Oct 25 20:00:26 2020 +0100
[FLINK-19802][connector files] Let BulkFormat createReader and restoreReader methods accept Splits directly
This is a step in supporting File Sources for BulkFormats that need subclasses of FileSourceSplit.
---
.../flink/connector/file/src/FileSource.java | 16 ++++-----
.../file/src/impl/FileRecordFormatAdapter.java | 23 ++++++++-----
.../connector/file/src/impl/FileSourceReader.java | 2 +-
.../file/src/impl/FileSourceSplitReader.java | 8 ++---
.../file/src/impl/StreamFormatAdapter.java | 27 +++++++--------
.../connector/file/src/reader/BulkFormat.java | 29 ++++++-----------
.../connector/file/src/impl/AdapterTestBase.java | 38 +++++++++++++---------
.../file/src/impl/FileRecordFormatAdapterTest.java | 3 +-
.../file/src/impl/StreamFormatAdapterTest.java | 5 +--
.../parquet/ParquetFileSystemFormatFactory.java | 3 +-
.../parquet/ParquetVectorizedInputFormat.java | 27 ++++++++-------
.../parquet/ParquetColumnarRowInputFormatTest.java | 16 +++++----
.../ParquetProtoStreamingFileSinkITCase.java | 1 -
.../table/connector/format/BulkDecodingFormat.java | 3 +-
.../table/factories/BulkReaderFormatFactory.java | 3 +-
.../table/filesystem/FileSystemTableSink.java | 7 ++--
.../table/filesystem/FileSystemTableSource.java | 9 ++---
.../stream/compact/CompactBulkReader.java | 15 ++++++---
.../filesystem/stream/compact/TestByteFormat.java | 4 ++-
19 files changed, 132 insertions(+), 107 deletions(-)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/FileSource.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/FileSource.java
index 36c61cd..6580742 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/FileSource.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/FileSource.java
@@ -140,7 +140,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp
private final FileSplitAssigner.Provider assignerFactory;
- private final BulkFormat<T> readerFormat;
+ private final BulkFormat<T, FileSourceSplit> readerFormat;
@Nullable
private final ContinuousEnumerationSettings continuousEnumerationSettings;
@@ -151,7 +151,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp
final Path[] inputPaths,
final FileEnumerator.Provider fileEnumerator,
final FileSplitAssigner.Provider splitAssigner,
- final BulkFormat<T> readerFormat,
+ final BulkFormat<T, FileSourceSplit> readerFormat,
@Nullable final ContinuousEnumerationSettings continuousEnumerationSettings) {
checkArgument(inputPaths.length > 0);
@@ -273,7 +273,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp
checkNotNull(paths, "paths");
checkArgument(paths.length > 0, "paths must not be empty");
- final BulkFormat<T> bulkFormat = new StreamFormatAdapter<>(reader);
+ final BulkFormat<T, FileSourceSplit> bulkFormat = new StreamFormatAdapter<>(reader);
return new FileSourceBuilder<>(paths, bulkFormat);
}
@@ -283,7 +283,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp
*
* <p>Examples for bulk readers are compressed and vectorized formats such as ORC or Parquet.
*/
- public static <T> FileSourceBuilder<T> forBulkFileFormat(final BulkFormat<T> reader, final Path... paths) {
+ public static <T> FileSourceBuilder<T> forBulkFileFormat(final BulkFormat<T, FileSourceSplit> reader, final Path... paths) {
checkNotNull(reader, "reader");
checkNotNull(paths, "paths");
checkArgument(paths.length > 0, "paths must not be empty");
@@ -303,7 +303,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp
checkNotNull(paths, "paths");
checkArgument(paths.length > 0, "paths must not be empty");
- final BulkFormat<T> bulkFormat = new FileRecordFormatAdapter<>(reader);
+ final BulkFormat<T, FileSourceSplit> bulkFormat = new FileRecordFormatAdapter<>(reader);
return new FileSourceBuilder<>(paths, bulkFormat);
}
@@ -322,7 +322,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp
* </ul>
*/
public static final class FileSourceBuilder<T> extends AbstractFileSourceBuilder<T, FileSourceBuilder<T>> {
- public FileSourceBuilder(Path[] inputPaths, BulkFormat<T> readerFormat) {
+ public FileSourceBuilder(Path[] inputPaths, BulkFormat<T, FileSourceSplit> readerFormat) {
super(inputPaths, readerFormat);
}
}
@@ -345,7 +345,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp
// mandatory - have no defaults
private final Path[] inputPaths;
- private final BulkFormat<T> readerFormat;
+ private final BulkFormat<T, FileSourceSplit> readerFormat;
// optional - have defaults
private FileEnumerator.Provider fileEnumerator;
@@ -353,7 +353,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp
@Nullable
private ContinuousEnumerationSettings continuousSourceSettings;
- protected AbstractFileSourceBuilder(Path[] inputPaths, BulkFormat<T> readerFormat) {
+ protected AbstractFileSourceBuilder(Path[] inputPaths, BulkFormat<T, FileSourceSplit> readerFormat) {
this.inputPaths = checkNotNull(inputPaths);
this.readerFormat = checkNotNull(readerFormat);
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileRecordFormatAdapter.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileRecordFormatAdapter.java
index e084926..22a7f72 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileRecordFormatAdapter.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileRecordFormatAdapter.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.file.src.impl;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.reader.FileRecordFormat;
import org.apache.flink.connector.file.src.util.CheckpointedPosition;
@@ -40,7 +41,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* The FormatReaderAdapter turns a {@link FileRecordFormat} into a {@link BulkFormat}.
*/
@Internal
-public final class FileRecordFormatAdapter<T> implements BulkFormat<T> {
+public final class FileRecordFormatAdapter<T> implements BulkFormat<T, FileSourceSplit> {
private static final long serialVersionUID = 1L;
@@ -53,11 +54,11 @@ public final class FileRecordFormatAdapter<T> implements BulkFormat<T> {
@Override
public BulkFormat.Reader<T> createReader(
final Configuration config,
- final Path filePath,
- final long splitOffset,
- final long splitLength) throws IOException {
+ final FileSourceSplit split) throws IOException {
+
+ final FileRecordFormat.Reader<T> reader =
+ fileFormat.createReader(config, split.path(), split.offset(), split.length());
- final FileRecordFormat.Reader<T> reader = fileFormat.createReader(config, filePath, splitOffset, splitLength);
return doWithCleanupOnException(reader, () -> {
//noinspection CodeBlock2Expr
return wrapReader(reader, config, CheckpointedPosition.NO_OFFSET, 0L);
@@ -67,10 +68,14 @@ public final class FileRecordFormatAdapter<T> implements BulkFormat<T> {
@Override
public BulkFormat.Reader<T> restoreReader(
final Configuration config,
- final Path filePath,
- final long splitOffset,
- final long splitLength,
- final CheckpointedPosition checkpointedPosition) throws IOException {
+ final FileSourceSplit split) throws IOException {
+
+ assert split.getReaderPosition().isPresent();
+ final CheckpointedPosition checkpointedPosition = split.getReaderPosition().get();
+
+ final Path filePath = split.path();
+ final long splitOffset = split.offset();
+ final long splitLength = split.length();
final FileRecordFormat.Reader<T> reader = checkpointedPosition.getOffset() == CheckpointedPosition.NO_OFFSET
? fileFormat.createReader(config, filePath, splitOffset, splitLength)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceReader.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceReader.java
index 8b492bb..1d7af6c 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceReader.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceReader.java
@@ -38,7 +38,7 @@ import java.util.Collection;
public final class FileSourceReader<T>
extends SingleThreadMultiplexSourceReaderBase<RecordAndPosition<T>, T, FileSourceSplit, FileSourceSplitState> {
- public FileSourceReader(SourceReaderContext readerContext, BulkFormat<T> readerFormat, Configuration config) {
+ public FileSourceReader(SourceReaderContext readerContext, BulkFormat<T, FileSourceSplit> readerFormat, Configuration config) {
super(
() -> new FileSourceSplitReader<>(config, readerFormat),
new FileSourceRecordEmitter<>(),
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceSplitReader.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceSplitReader.java
index ed9159b..b3495f9 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceSplitReader.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceSplitReader.java
@@ -48,7 +48,7 @@ final class FileSourceSplitReader<T> implements SplitReader<RecordAndPosition<T>
private static final Logger LOG = LoggerFactory.getLogger(FileSourceSplitReader.class);
private final Configuration config;
- private final BulkFormat<T> readerFactory;
+ private final BulkFormat<T, FileSourceSplit> readerFactory;
private final Queue<FileSourceSplit> splits;
@@ -57,7 +57,7 @@ final class FileSourceSplitReader<T> implements SplitReader<RecordAndPosition<T>
@Nullable
private String currentSplitId;
- public FileSourceSplitReader(Configuration config, BulkFormat<T> readerFactory) {
+ public FileSourceSplitReader(Configuration config, BulkFormat<T, FileSourceSplit> readerFactory) {
this.config = config;
this.readerFactory = readerFactory;
this.splits = new ArrayDeque<>();
@@ -99,8 +99,8 @@ final class FileSourceSplitReader<T> implements SplitReader<RecordAndPosition<T>
final Optional<CheckpointedPosition> position = nextSplit.getReaderPosition();
currentReader = position.isPresent()
- ? readerFactory.restoreReader(config, nextSplit.path(), nextSplit.offset(), nextSplit.length(), position.get())
- : readerFactory.createReader(config, nextSplit.path(), nextSplit.offset(), nextSplit.length());
+ ? readerFactory.restoreReader(config, nextSplit)
+ : readerFactory.createReader(config, nextSplit);
}
private FileRecords<T> finishSplit() throws IOException {
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
index e085a0e..6bf2ff2 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.reader.StreamFormat;
import org.apache.flink.connector.file.src.util.CheckpointedPosition;
@@ -45,7 +46,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* Adapter to turn a {@link StreamFormat} into a {@link BulkFormat}.
*/
@Internal
-public final class StreamFormatAdapter<T> implements BulkFormat<T> {
+public final class StreamFormatAdapter<T> implements BulkFormat<T, FileSourceSplit> {
private static final long serialVersionUID = 1L;
@@ -58,15 +59,14 @@ public final class StreamFormatAdapter<T> implements BulkFormat<T> {
@Override
public BulkFormat.Reader<T> createReader(
final Configuration config,
- final Path filePath,
- final long splitOffset,
- final long splitLength) throws IOException {
+ final FileSourceSplit split) throws IOException {
- final TrackingFsDataInputStream trackingStream = openStream(filePath, config, splitOffset);
+ final TrackingFsDataInputStream trackingStream = openStream(split.path(), config, split.offset());
+ final long splitEnd = split.offset() + split.length();
return doWithCleanupOnException(trackingStream, () -> {
final StreamFormat.Reader<T> streamReader = streamFormat.createReader(
- config, trackingStream, trackingStream.getFileLength(), splitOffset + splitLength);
+ config, trackingStream, trackingStream.getFileLength(), splitEnd);
return new Reader<>(streamReader, trackingStream, CheckpointedPosition.NO_OFFSET, 0L);
});
}
@@ -74,22 +74,23 @@ public final class StreamFormatAdapter<T> implements BulkFormat<T> {
@Override
public BulkFormat.Reader<T> restoreReader(
final Configuration config,
- final Path filePath,
- final long splitOffset,
- final long splitLength,
- final CheckpointedPosition checkpointedPosition) throws IOException {
+ final FileSourceSplit split) throws IOException {
- final TrackingFsDataInputStream trackingStream = openStream(filePath, config, splitOffset);
+ assert split.getReaderPosition().isPresent();
+ final CheckpointedPosition checkpointedPosition = split.getReaderPosition().get();
+
+ final TrackingFsDataInputStream trackingStream = openStream(split.path(), config, split.offset());
+ final long splitEnd = split.offset() + split.length();
return doWithCleanupOnException(trackingStream, () -> {
// if there never was a checkpointed offset, yet, we need to initialize the reader like a fresh reader.
// see the JavaDocs on StreamFormat.restoreReader() for details
final StreamFormat.Reader<T> streamReader = checkpointedPosition.getOffset() == CheckpointedPosition.NO_OFFSET
? streamFormat.createReader(
- config, trackingStream, trackingStream.getFileLength(), splitOffset + splitLength)
+ config, trackingStream, trackingStream.getFileLength(), splitEnd)
: streamFormat.restoreReader(
config, trackingStream, checkpointedPosition.getOffset(),
- trackingStream.getFileLength(), splitOffset + splitLength);
+ trackingStream.getFileLength(), splitEnd);
// skip the records to skip, but make sure we close the reader if something goes wrong
doWithCleanupOnException(streamReader, () -> {
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/reader/BulkFormat.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/reader/BulkFormat.java
index 5c378ed..cc37b95 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/reader/BulkFormat.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/reader/BulkFormat.java
@@ -22,10 +22,9 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.util.MutableRecordAndPosition;
import org.apache.flink.connector.file.src.util.RecordAndPosition;
-import org.apache.flink.core.fs.Path;
import javax.annotation.Nullable;
@@ -39,9 +38,9 @@ import java.io.Serializable;
*
* <p>The outer {@code 'BulkFormat'} class acts mainly as a configuration holder and factory for the reader.
* The actual reading is done by the {@link BulkFormat.Reader}, which is created in the
- * {@link BulkFormat#createReader(Configuration, Path, long, long)} method. If a bulk reader is created
+ * {@link BulkFormat#createReader(Configuration, FileSourceSplit)} method. If a bulk reader is created
* based on a checkpoint during checkpointed streaming execution, then the reader is re-created in
- * the {@link BulkFormat#restoreReader(Configuration, Path, long, long, CheckpointedPosition)} method.
+ * the {@link BulkFormat#restoreReader(Configuration, FileSourceSplit)} method.
*
* <h2>Splitting</h2>
*
@@ -92,30 +91,22 @@ import java.io.Serializable;
* handed over as one.
*/
@PublicEvolving
-public interface BulkFormat<T> extends Serializable, ResultTypeQueryable<T> {
+public interface BulkFormat<T, SplitT extends FileSourceSplit> extends Serializable, ResultTypeQueryable<T> {
/**
- * Creates a new reader that reads from {@code filePath} starting at {@code offset} and reads
- * until {@code length} bytes after the offset.
+ * Creates a new reader that reads from the {@link FileSourceSplit#path() split's path}
+ * starting at the {@link FileSourceSplit#offset()} split's offset} and reads
+ * {@link FileSourceSplit#length() length} bytes after the offset.
*/
- BulkFormat.Reader<T> createReader(
- Configuration config,
- Path filePath,
- long splitOffset,
- long splitLength) throws IOException;
+ BulkFormat.Reader<T> createReader(Configuration config, SplitT split) throws IOException;
/**
- * Creates a new reader that reads from {@code filePath} starting at {@code offset} and reads
+ * Creates a new reader that reads from {@code split.path()} starting at {@code offset} and reads
* until {@code length} bytes after the offset. A number of {@code recordsToSkip} records should be
* read and discarded after the offset. This is typically part of restoring a reader to a checkpointed
* position.
*/
- BulkFormat.Reader<T> restoreReader(
- Configuration config,
- Path filePath,
- long splitOffset,
- long splitLength,
- CheckpointedPosition checkpointedPosition) throws IOException;
+ BulkFormat.Reader<T> restoreReader(Configuration config, SplitT split) throws IOException;
/**
* Checks whether this format is splittable. Splittable formats allow Flink to create multiple splits
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/AdapterTestBase.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/AdapterTestBase.java
index ea59379..eee99e1 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/AdapterTestBase.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/AdapterTestBase.java
@@ -18,7 +18,6 @@
package org.apache.flink.connector.file.src.impl;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.file.src.FileSourceSplit;
@@ -85,7 +84,7 @@ public abstract class AdapterTestBase<FormatT> {
protected abstract FormatT createFormatFailingInInstantiation();
- protected abstract BulkFormat<Integer> wrapWithAdapter(FormatT format);
+ protected abstract BulkFormat<Integer, FileSourceSplit> wrapWithAdapter(FormatT format);
// ------------------------------------------------------------------------
// shared tests
@@ -115,7 +114,7 @@ public abstract class AdapterTestBase<FormatT> {
final Configuration config = new Configuration();
config.set(StreamFormat.FETCH_IO_SIZE, new MemorySize(10));
- final BulkFormat<Integer> adapter = wrapWithAdapter(format);
+ final BulkFormat<Integer, FileSourceSplit> adapter = wrapWithAdapter(format);
final Queue<FileSourceSplit> splits = buildSplits(numSplits);
final List<Integer> result = new ArrayList<>();
@@ -123,14 +122,14 @@ public abstract class AdapterTestBase<FormatT> {
BulkFormat.Reader<Integer> currentReader = null;
for (int nextRecordToRecover : boundaries) {
- final Tuple2<FileSourceSplit, CheckpointedPosition> toRecoverFrom = readNumbers(
+ final FileSourceSplit toRecoverFrom = readNumbers(
currentReader, currentSplit,
adapter, splits, config,
result,
nextRecordToRecover - result.size());
- currentSplit = toRecoverFrom.f0;
- currentReader = adapter.restoreReader(config, currentSplit.path(), currentSplit.offset(), currentSplit.length(), toRecoverFrom.f1);
+ currentSplit = toRecoverFrom;
+ currentReader = toRecoverFrom == null ? null : adapter.restoreReader(config, toRecoverFrom);
}
verifyIntListResult(result);
@@ -148,9 +147,9 @@ public abstract class AdapterTestBase<FormatT> {
testFs.register();
// test
- final BulkFormat<Integer> adapter = wrapWithAdapter(createFormatFailingInInstantiation());
+ final BulkFormat<Integer, FileSourceSplit> adapter = wrapWithAdapter(createFormatFailingInInstantiation());
try {
- adapter.createReader(new Configuration(), testPath, 0, 1024);
+ adapter.createReader(new Configuration(), new FileSourceSplit("id", testPath, 0, 1024));
} catch (IOException ignored) {}
// assertions
@@ -170,11 +169,16 @@ public abstract class AdapterTestBase<FormatT> {
testFs.register();
// test
- final BulkFormat<Integer> adapter = wrapWithAdapter(createFormatFailingInInstantiation());
+ final BulkFormat<Integer, FileSourceSplit> adapter = wrapWithAdapter(createFormatFailingInInstantiation());
+ final FileSourceSplit split = new FileSourceSplit(
+ "id",
+ testPath,
+ 0, 1024,
+ new String[0],
+ new CheckpointedPosition(0L, 5L));
+
try {
- adapter.restoreReader(
- new Configuration(), testPath, 0, 1024,
- new CheckpointedPosition(0L, 5L));
+ adapter.restoreReader(new Configuration(), split);
} catch (IOException ignored) {}
// assertions
@@ -202,10 +206,10 @@ public abstract class AdapterTestBase<FormatT> {
readNumbers(reader, null, null, null, null, result, num);
}
- protected static Tuple2<FileSourceSplit, CheckpointedPosition> readNumbers(
+ protected static FileSourceSplit readNumbers(
BulkFormat.Reader<Integer> currentReader,
FileSourceSplit currentSplit,
- BulkFormat<Integer> format,
+ BulkFormat<Integer, FileSourceSplit> format,
Queue<FileSourceSplit> moreSplits,
Configuration config,
List<Integer> result,
@@ -219,7 +223,7 @@ public abstract class AdapterTestBase<FormatT> {
if (currentReader == null) {
currentSplit = moreSplits.poll();
assertNotNull(currentSplit);
- currentReader = format.createReader(config, currentSplit.path(), currentSplit.offset(), currentSplit.length());
+ currentReader = format.createReader(config, currentSplit);
}
// loop across batches
@@ -240,7 +244,9 @@ public abstract class AdapterTestBase<FormatT> {
currentReader = null;
}
- return new Tuple2<>(currentSplit, new CheckpointedPosition(offset, skip));
+ return currentSplit != null
+ ? currentSplit.updateWithCheckpointedPosition(new CheckpointedPosition(offset, skip))
+ : null;
}
static Queue<FileSourceSplit> buildSplits(int numSplits) {
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/FileRecordFormatAdapterTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/FileRecordFormatAdapterTest.java
index 779f38e..888b83e 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/FileRecordFormatAdapterTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/FileRecordFormatAdapterTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.file.src.impl;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.reader.FileRecordFormat;
import org.apache.flink.core.fs.FSDataInputStream;
@@ -56,7 +57,7 @@ public class FileRecordFormatAdapterTest extends AdapterTestBase<FileRecordForma
}
@Override
- protected BulkFormat<Integer> wrapWithAdapter(FileRecordFormat<Integer> format) {
+ protected BulkFormat<Integer, FileSourceSplit> wrapWithAdapter(FileRecordFormat<Integer> format) {
return new FileRecordFormatAdapter<>(format);
}
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapterTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapterTest.java
index 29e74b1..c336a1a 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapterTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapterTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.reader.SimpleStreamFormat;
import org.apache.flink.connector.file.src.reader.StreamFormat;
@@ -61,7 +62,7 @@ public class StreamFormatAdapterTest extends AdapterTestBase<StreamFormat<Intege
}
@Override
- protected BulkFormat<Integer> wrapWithAdapter(StreamFormat<Integer> format) {
+ protected BulkFormat<Integer, FileSourceSplit> wrapWithAdapter(StreamFormat<Integer> format) {
return new StreamFormatAdapter<>(format);
}
@@ -88,7 +89,7 @@ public class StreamFormatAdapterTest extends AdapterTestBase<StreamFormat<Intege
final Configuration config = new Configuration();
config.set(StreamFormat.FETCH_IO_SIZE, new MemorySize(batchSize));
final StreamFormatAdapter<Integer> format = new StreamFormatAdapter<>(new CheckpointedIntFormat());
- final BulkFormat.Reader<Integer> reader = format.createReader(config, testPath, 0L, FILE_LEN);
+ final BulkFormat.Reader<Integer> reader = format.createReader(config, new FileSourceSplit("test-id", testPath, 0L, FILE_LEN));
final List<Integer> result = new ArrayList<>();
readNumbers(reader, result, NUM_NUMBERS);
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileSystemFormatFactory.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileSystemFormatFactory.java
index ec4c95d..75d3000 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileSystemFormatFactory.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileSystemFormatFactory.java
@@ -21,6 +21,7 @@ package org.apache.flink.formats.parquet;
import org.apache.flink.api.common.serialization.BulkWriter;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.formats.parquet.row.ParquetRowDataBuilder;
import org.apache.flink.table.connector.ChangelogMode;
@@ -65,7 +66,7 @@ public class ParquetFileSystemFormatFactory implements BulkReaderFormatFactory,
DynamicTableFactory.Context context, ReadableConfig formatOptions) {
return new BulkDecodingFormat<RowData>() {
@Override
- public BulkFormat<RowData> createRuntimeDecoder(
+ public BulkFormat<RowData, FileSourceSplit> createRuntimeDecoder(
DynamicTableSource.Context sourceContext,
DataType producedDataType) {
return ParquetColumnarRowInputFormat.createPartitionedFormat(
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
index ab39048..1f67058 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
@@ -20,6 +20,7 @@ package org.apache.flink.formats.parquet;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.util.CheckpointedPosition;
import org.apache.flink.connector.file.src.util.Pool;
@@ -69,7 +70,7 @@ import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter;
* Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
* vectorized mode.
*/
-public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T, FileSourceSplit> {
private static final long serialVersionUID = 1L;
@@ -99,10 +100,13 @@ public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
@Override
public ParquetReader createReader(
- Configuration config,
- Path filePath,
- long splitOffset,
- long splitLength) throws IOException {
+ final Configuration config,
+ final FileSourceSplit split) throws IOException {
+
+ final Path filePath = split.path();
+ final long splitOffset = split.offset();
+ final long splitLength = split.length();
+
org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
ParquetMetadata footer = readFooter(
hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
@@ -134,15 +138,16 @@ public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
@Override
public ParquetReader restoreReader(
- Configuration config,
- Path filePath,
- long splitOffset,
- long splitLength,
- CheckpointedPosition checkpointedPosition) throws IOException {
+ final Configuration config,
+ final FileSourceSplit split) throws IOException {
+
+ assert split.getReaderPosition().isPresent();
+ final CheckpointedPosition checkpointedPosition = split.getReaderPosition().get();
+
Preconditions.checkArgument(
checkpointedPosition.getOffset() == CheckpointedPosition.NO_OFFSET,
"The offset of CheckpointedPosition should always be NO_OFFSET");
- ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ ParquetReader reader = createReader(config, split);
reader.seek(checkpointedPosition.getRecordsAfterOffset());
return reader;
}
diff --git a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java
index d3fe757..3986f3f 100644
--- a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java
+++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.formats.parquet;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.util.CheckpointedPosition;
import org.apache.flink.core.fs.Path;
@@ -225,7 +226,7 @@ public class ParquetColumnarRowInputFormatTest {
true);
AtomicInteger cnt = new AtomicInteger(0);
- forEachRemaining(format.createReader(EMPTY_CONF, testPath, 0, Long.MAX_VALUE), row -> {
+ forEachRemaining(format.createReader(EMPTY_CONF, new FileSourceSplit("id", testPath, 0, Long.MAX_VALUE)), row -> {
int i = cnt.get();
assertEquals(i, row.getDouble(0), 0);
assertEquals((byte) i, row.getByte(1));
@@ -352,10 +353,13 @@ public class ParquetColumnarRowInputFormatTest {
BulkFormat.Reader<RowData> reader = format.restoreReader(
EMPTY_CONF,
- path,
- splitStart,
- splitLength,
- new CheckpointedPosition(CheckpointedPosition.NO_OFFSET, seekToRow));
+ new FileSourceSplit(
+ "id",
+ path,
+ splitStart,
+ splitLength,
+ new String[0],
+ new CheckpointedPosition(CheckpointedPosition.NO_OFFSET, seekToRow)));
AtomicInteger cnt = new AtomicInteger(0);
forEachRemaining(reader, row -> {
@@ -483,7 +487,7 @@ public class ParquetColumnarRowInputFormatTest {
true);
AtomicInteger cnt = new AtomicInteger(0);
- forEachRemaining(format.createReader(EMPTY_CONF, testPath, 0, Long.MAX_VALUE), row -> {
+ forEachRemaining(format.createReader(EMPTY_CONF, new FileSourceSplit("id", testPath, 0, Long.MAX_VALUE)), row -> {
int i = cnt.get();
// common values
assertEquals(i, row.getDouble(0), 0);
diff --git a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoStreamingFileSinkITCase.java b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoStreamingFileSinkITCase.java
index 114be739..27a5ba6 100644
--- a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoStreamingFileSinkITCase.java
+++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoStreamingFileSinkITCase.java
@@ -50,7 +50,6 @@ import static org.junit.Assert.assertTrue;
* Simple integration test case for writing bulk encoded files with the
* {@link StreamingFileSink} with Parquet.
*/
-@SuppressWarnings("serial")
public class ParquetProtoStreamingFileSinkITCase extends AbstractTestBase {
@Rule
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/format/BulkDecodingFormat.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/format/BulkDecodingFormat.java
index de1a186..7619451 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/format/BulkDecodingFormat.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/format/BulkDecodingFormat.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.connector.format;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.expressions.ResolvedExpression;
@@ -29,7 +30,7 @@ import java.util.List;
* A {@link Format} for a {@link DynamicTableSource} for reading rows by {@link BulkFormat}.
*/
@Internal
-public interface BulkDecodingFormat<T> extends DecodingFormat<BulkFormat<T>> {
+public interface BulkDecodingFormat<T> extends DecodingFormat<BulkFormat<T, FileSourceSplit>> {
/**
* Provides the expected maximum number of produced records for limiting on a best-effort basis.
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/BulkReaderFormatFactory.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/BulkReaderFormatFactory.java
index 5f27d1f..fcb05d4 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/BulkReaderFormatFactory.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/BulkReaderFormatFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.factories;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.table.connector.format.BulkDecodingFormat;
import org.apache.flink.table.data.RowData;
@@ -30,7 +31,7 @@ import org.apache.flink.table.data.RowData;
* @see FactoryUtil#createTableFactoryHelper(DynamicTableFactory, DynamicTableFactory.Context)
*/
@Internal
-public interface BulkReaderFormatFactory extends DecodingFormatFactory<BulkFormat<RowData>> {
+public interface BulkReaderFormatFactory extends DecodingFormatFactory<BulkFormat<RowData, FileSourceSplit>> {
/**
* Creates a {@link BulkDecodingFormat} from the given context and format options.
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSink.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSink.java
index 2b22fe0..f0fe98b 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSink.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSink.java
@@ -27,6 +27,7 @@ import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.core.fs.FSDataOutputStream;
import org.apache.flink.core.fs.FileSystem;
@@ -94,7 +95,7 @@ public class FileSystemTableSink extends AbstractFileSystemTable implements
SupportsPartitioning,
SupportsOverwrite {
- @Nullable private final DecodingFormat<BulkFormat<RowData>> bulkReaderFormat;
+ @Nullable private final DecodingFormat<BulkFormat<RowData, FileSourceSplit>> bulkReaderFormat;
@Nullable private final EncodingFormat<BulkWriter.Factory<RowData>> bulkWriterFormat;
@Nullable private final EncodingFormat<SerializationSchema<RowData>> serializationFormat;
@Nullable private final FileSystemFormatFactory formatFactory;
@@ -105,7 +106,7 @@ public class FileSystemTableSink extends AbstractFileSystemTable implements
FileSystemTableSink(
DynamicTableFactory.Context context,
- @Nullable DecodingFormat<BulkFormat<RowData>> bulkReaderFormat,
+ @Nullable DecodingFormat<BulkFormat<RowData, FileSourceSplit>> bulkReaderFormat,
@Nullable EncodingFormat<BulkWriter.Factory<RowData>> bulkWriterFormat,
@Nullable EncodingFormat<SerializationSchema<RowData>> serializationFormat,
@Nullable FileSystemFormatFactory formatFactory) {
@@ -241,7 +242,7 @@ public class FileSystemTableSink extends AbstractFileSystemTable implements
private Optional<CompactReader.Factory<RowData>> createCompactReaderFactory(Context context) {
DataType producedDataType = schema.toRowDataType();
if (bulkReaderFormat != null) {
- BulkFormat<RowData> format = bulkReaderFormat.createRuntimeDecoder(
+ BulkFormat<RowData, FileSourceSplit> format = bulkReaderFormat.createRuntimeDecoder(
createSourceContext(context), producedDataType);
return Optional.of(CompactBulkReader.factory(format));
} else if (formatFactory != null) {
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSource.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSource.java
index f0bbe37..1dfbb75 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSource.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSource.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.java.io.CollectionInputFormat;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.file.src.FileSource;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
@@ -73,7 +74,7 @@ public class FileSystemTableSource extends AbstractFileSystemTable implements
SupportsPartitionPushDown,
SupportsFilterPushDown {
- @Nullable private final DecodingFormat<BulkFormat<RowData>> bulkReaderFormat;
+ @Nullable private final DecodingFormat<BulkFormat<RowData, FileSourceSplit>> bulkReaderFormat;
@Nullable private final DecodingFormat<DeserializationSchema<RowData>> deserializationFormat;
@Nullable private final FileSystemFormatFactory formatFactory;
@@ -84,7 +85,7 @@ public class FileSystemTableSource extends AbstractFileSystemTable implements
public FileSystemTableSource(
DynamicTableFactory.Context context,
- @Nullable DecodingFormat<BulkFormat<RowData>> bulkReaderFormat,
+ @Nullable DecodingFormat<BulkFormat<RowData, FileSourceSplit>> bulkReaderFormat,
@Nullable DecodingFormat<DeserializationSchema<RowData>> deserializationFormat,
@Nullable FileSystemFormatFactory formatFactory) {
super(context);
@@ -124,7 +125,7 @@ public class FileSystemTableSource extends AbstractFileSystemTable implements
}
private SourceProvider sourceProvider(
- DecodingFormat<BulkFormat<RowData>> decodingFormat, ScanContext scanContext) {
+ DecodingFormat<BulkFormat<RowData, FileSourceSplit>> decodingFormat, ScanContext scanContext) {
if (decodingFormat instanceof BulkDecodingFormat) {
BulkDecodingFormat<RowData> bulkFormat = (BulkDecodingFormat<RowData>) decodingFormat;
if (limit != null) {
@@ -134,7 +135,7 @@ public class FileSystemTableSource extends AbstractFileSystemTable implements
bulkFormat.applyFilters(filters);
}
}
- BulkFormat<RowData> bulkFormat = decodingFormat.createRuntimeDecoder(
+ BulkFormat<RowData, FileSourceSplit> bulkFormat = decodingFormat.createRuntimeDecoder(
scanContext, getProducedDataType());
FileSource.FileSourceBuilder<RowData> builder = FileSource
.forBulkFileFormat(bulkFormat, paths());
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/stream/compact/CompactBulkReader.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/stream/compact/CompactBulkReader.java
index 670bd55..01498ec 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/stream/compact/CompactBulkReader.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/stream/compact/CompactBulkReader.java
@@ -18,10 +18,12 @@
package org.apache.flink.table.filesystem.stream.compact;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.util.RecordAndPosition;
import java.io.IOException;
+import java.util.UUID;
/**
* The {@link CompactReader} to delegate {@link CompactBulkReader}.
@@ -57,7 +59,7 @@ public class CompactBulkReader<T> implements CompactReader<T> {
reader.close();
}
- public static <T> CompactReader.Factory<T> factory(BulkFormat<T> format) {
+ public static <T> CompactReader.Factory<T> factory(BulkFormat<T, FileSourceSplit> format) {
return new Factory<>(format);
}
@@ -66,17 +68,20 @@ public class CompactBulkReader<T> implements CompactReader<T> {
*/
private static class Factory<T> implements CompactReader.Factory<T> {
- private final BulkFormat<T> format;
+ private static final long serialVersionUID = 1L;
- public Factory(BulkFormat<T> format) {
+ private final BulkFormat<T, FileSourceSplit> format;
+
+ public Factory(BulkFormat<T, FileSourceSplit> format) {
this.format = format;
}
@Override
public CompactReader<T> create(CompactContext context) throws IOException {
- long len = context.getFileSystem().getFileStatus(context.getPath()).getLen();
+ final String splitId = UUID.randomUUID().toString();
+ final long len = context.getFileSystem().getFileStatus(context.getPath()).getLen();
return new CompactBulkReader<>(format.createReader(
- context.getConfig(), context.getPath(), 0, len));
+ context.getConfig(), new FileSourceSplit(splitId, context.getPath(), 0, len)));
}
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/filesystem/stream/compact/TestByteFormat.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/filesystem/stream/compact/TestByteFormat.java
index e306462..1dc440f 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/filesystem/stream/compact/TestByteFormat.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/filesystem/stream/compact/TestByteFormat.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.filesystem.stream.compact;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.impl.StreamFormatAdapter;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.reader.SimpleStreamFormat;
@@ -31,6 +32,7 @@ import java.io.IOException;
/**
* Test byte stream format.
*/
+@SuppressWarnings("serial")
public class TestByteFormat extends SimpleStreamFormat<Byte> {
@Override
@@ -58,7 +60,7 @@ public class TestByteFormat extends SimpleStreamFormat<Byte> {
return BasicTypeInfo.BYTE_TYPE_INFO;
}
- public static BulkFormat<Byte> bulkFormat() {
+ public static BulkFormat<Byte, FileSourceSplit> bulkFormat() {
return new StreamFormatAdapter<>(new TestByteFormat());
}
}