Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1895#discussion_r60914650
--- Diff:
flink-core/src/main/java/org/apache/flink/api/common/io/BinaryInputFormat.java
---
@@ -40,36 +41,46 @@
import java.util.List;
/**
- * Base class for all input formats that use blocks of fixed size. The
input splits are aligned to these blocks. Without
- * configuration, these block sizes equal the native block sizes of the
HDFS.
+ * Base class for all input formats that use blocks of fixed size. The
input splits are aligned to these blocks,
+ * meaning that each split will consist of one block. Without
configuration, these block sizes equal the native
+ * block sizes of the HDFS.
+ *
+ * A block will contain a {@link BlockInfo} at the end of the block.
There, the reader can find some statistics
+ * about the split currently being read, that will help correctly parse
the contents of the block.
*/
@Public
-public abstract class BinaryInputFormat<T> extends FileInputFormat<T> {
+public abstract class BinaryInputFormat<T> extends FileInputFormat<T>
+ implements CheckpointableInputFormat<FileInputSplit, Tuple3<Boolean,
Long, Long>> {
+
private static final long serialVersionUID = 1L;
- /**
- * The log.
- */
+ /** The log. */
private static final Logger LOG =
LoggerFactory.getLogger(BinaryInputFormat.class);
- /**
- * The config parameter which defines the fixed length of a record.
- */
+ /** The config parameter which defines the fixed length of a record. */
public static final String BLOCK_SIZE_PARAMETER_KEY =
"input.block_size";
public static final long NATIVE_BLOCK_SIZE = Long.MIN_VALUE;
- /**
- * The block size to use.
- */
+ /** The block size to use. */
private long blockSize = NATIVE_BLOCK_SIZE;
private transient DataInputViewStreamWrapper dataInputStream;
+ /** The BlockInfo for the Block corresponding to the split currently
being read. */
private transient BlockInfo blockInfo;
+ /** A wrapper around the block currently being read. */
+ private transient BlockBasedInput blockBasedInput = null;
+
+ /**
+ * The number of records already read from the block.
+ * This is used to decide if the end of the block has been
+ * reached.
+ */
private long readRecords;
+// private transient Tuple3<Boolean, Long, Long> restoredState;
--- End diff --
Probably left over from the previous version of the code
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
