import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public StateTable<K, N, SV> deserialize(
		String stateName,
		HeapKeyedStateBackend<K> stateBackend) throws IOException {

	final FileSystem fs = getFilePath().getFileSystem();
	try (FSDataInputStream inStream = fs.open(getFilePath())) {
		final DataInputViewStreamWrapper inView = new DataInputViewStreamWrapper(inStream);
		AbstractMigrationRestoreStrategy<K, N, SV> restoreStrategy =
				new AbstractMigrationRestoreStrategy<K, N, SV>(keySerializer, namespaceSerializer, stateSerializer) {
					@Override
					protected DataInputView openDataInputView() throws IOException {
						return inView;
					}
				};
		return restoreStrategy.deserialize(stateName, stateBackend);
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@Override
public void open() throws Exception {
	super.open();

	if (serializedInitialValue == null) {
		throw new RuntimeException("No initial value was serialized for the fold " +
				"operator. Probably the setOutputType method was not called.");
	}

	try (ByteArrayInputStream bais = new ByteArrayInputStream(serializedInitialValue);
		DataInputViewStreamWrapper in = new DataInputViewStreamWrapper(bais)) {
		initialValue = outTypeSerializer.deserialize(in);
	}

	ValueStateDescriptor<OUT> stateId = new ValueStateDescriptor<>(STATE_NAME, outTypeSerializer);
	values = getPartitionedState(stateId);
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
private void restoreState() throws Exception {
    LOGGER.info("Restore siddhi state");
    final Iterator<byte[]> siddhiState = siddhiRuntimeState.get().iterator();
    if (siddhiState.hasNext()) {
        this.siddhiRuntime.restore(siddhiState.next());
    }

    LOGGER.info("Restore queued records state");
    final Iterator<byte[]> queueState = queuedRecordsState.get().iterator();
    if (queueState.hasNext()) {
        final ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(queueState.next());
        final DataInputViewStreamWrapper dataInputView = new DataInputViewStreamWrapper(byteArrayInputStream);
        try {
            this.priorityQueue = restoreQueuerState(dataInputView);
        } finally {
            dataInputView.close();
            byteArrayInputStream.close();
        }
    }
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
public void restoreStateForKeyGroup(DataInputViewStreamWrapper stream, int keyGroupIdx,
									ClassLoader userCodeClassLoader) throws IOException, ClassNotFoundException {

	int noOfTimerServices = stream.readInt();
	for (int i = 0; i < noOfTimerServices; i++) {
		String serviceName = stream.readUTF();

		HeapInternalTimerService<K, N> timerService = timerServices.get(serviceName);
		if (timerService == null) {
			timerService = new HeapInternalTimerService<>(
				totalKeyGroups,
				localKeyGroupRange,
				keyContext,
				processingTimeService);
			timerServices.put(serviceName, timerService);
		}
		timerService.restoreTimersForKeyGroup(stream, keyGroupIdx, userCodeClassLoader);
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@Test
public void getKeyedStateStreams() throws Exception {

	int readKeyGroupCount = 0;

	for (KeyGroupStatePartitionStreamProvider stateStreamProvider
			: initializationContext.getRawKeyedStateInputs()) {

		Assert.assertNotNull(stateStreamProvider);

		try (InputStream is = stateStreamProvider.getStream()) {
			DataInputView div = new DataInputViewStreamWrapper(is);
			int val = div.readInt();
			++readKeyGroupCount;
			Assert.assertEquals(stateStreamProvider.getKeyGroupId(), val);
		}
	}

	Assert.assertEquals(writtenKeyGroups, readKeyGroupCount);
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
private static void validateDeserialization(TypeSerializer<User> serializer) throws IOException {
	final Random rnd = new Random(RANDOM_SEED);

	try (InputStream in = BackwardsCompatibleAvroSerializerTest.class.getClassLoader()
			.getResourceAsStream(DATA_RESOURCE)) {

		final DataInputViewStreamWrapper inView = new DataInputViewStreamWrapper(in);

		for (int i = 0; i < NUM_DATA_ENTRIES; i++) {
			final User deserialized = serializer.deserialize(inView);

			// deterministically generate a reference record
			final User reference = TestDataGenerator.generateRandomUser(rnd);

			assertEquals(reference, deserialized);
		}
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@Test
public void getOperatorStateStreams() throws Exception {

	int i = 0;
	int s = 0;
	for (StatePartitionStreamProvider streamProvider : initializationContext.getRawOperatorStateInputs()) {
		if (0 == i % 4) {
			++i;
		}
		Assert.assertNotNull(streamProvider);
		try (InputStream is = streamProvider.getStream()) {
			DataInputView div = new DataInputViewStreamWrapper(is);

			int val = div.readInt();
			Assert.assertEquals(i * NUM_HANDLES + s, val);
		}

		++s;
		if (s == i % 4) {
			s = 0;
			++i;
		}
	}

}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@Override
public Iterable<V> get() {
	try {
		writeCurrentKeyWithGroupAndNamespace();
		byte[] key = keySerializationStream.toByteArray();
		byte[] valueBytes = backend.db.get(columnFamily, key);

		if (valueBytes == null) {
			return null;
		}

		ByteArrayInputStream bais = new ByteArrayInputStream(valueBytes);
		DataInputViewStreamWrapper in = new DataInputViewStreamWrapper(bais);

		List<V> result = new ArrayList<>();
		while (in.available() > 0) {
			result.add(valueSerializer.deserialize(in));
			if (in.available() > 0) {
				in.readByte();
			}
		}
		return result;
	} catch (IOException | RocksDBException e) {
		throw new RuntimeException("Error while retrieving data from RocksDB", e);
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@Override
public void add(V value) throws IOException {
	try {
		writeCurrentKeyWithGroupAndNamespace();
		byte[] key = keySerializationStream.toByteArray();
		byte[] valueBytes = backend.db.get(columnFamily, key);

		DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(keySerializationStream);
		if (valueBytes == null) {
			keySerializationStream.reset();
			valueSerializer.serialize(value, out);
			backend.db.put(columnFamily, writeOptions, key, keySerializationStream.toByteArray());
		} else {
			V oldValue = valueSerializer.deserialize(new DataInputViewStreamWrapper(new ByteArrayInputStream(valueBytes)));
			V newValue = reduceFunction.reduce(oldValue, value);
			keySerializationStream.reset();
			valueSerializer.serialize(newValue, out);
			backend.db.put(columnFamily, writeOptions, key, keySerializationStream.toByteArray());
		}
	} catch (Exception e) {
		throw new RuntimeException("Error while adding data to RocksDB", e);
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@Override
public void add(T value) throws IOException {
	try {
		writeCurrentKeyWithGroupAndNamespace();
		byte[] key = keySerializationStream.toByteArray();
		byte[] valueBytes = backend.db.get(columnFamily, key);
		DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(keySerializationStream);
		if (valueBytes == null) {
			keySerializationStream.reset();
			valueSerializer.serialize(foldFunction.fold(stateDesc.getDefaultValue(), value), out);
			backend.db.put(columnFamily, writeOptions, key, keySerializationStream.toByteArray());
		} else {
			ACC oldValue = valueSerializer.deserialize(new DataInputViewStreamWrapper(new ByteArrayInputStreamWithPos(valueBytes)));
			ACC newValue = foldFunction.fold(oldValue, value);
			keySerializationStream.reset();
			valueSerializer.serialize(newValue, out);
			backend.db.put(columnFamily, writeOptions, key, keySerializationStream.toByteArray());
		}
	} catch (Exception e) {
		throw new RuntimeException("Error while adding data to RocksDB", e);
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
/**
 * Restore one key groups state handle.
 */
private void restoreKeyGroupsInStateHandle()
	throws IOException, StateMigrationException, RocksDBException {
	try {
		currentStateHandleInStream = currentKeyGroupsStateHandle.openInputStream();
		rocksDBKeyedStateBackend.cancelStreamRegistry.registerCloseable(currentStateHandleInStream);
		currentStateHandleInView = new DataInputViewStreamWrapper(currentStateHandleInStream);
		restoreKVStateMetaData();
		restoreKVStateData();
	} finally {
		if (currentStateHandleInStream != null
			&& rocksDBKeyedStateBackend.cancelStreamRegistry.unregisterCloseable(currentStateHandleInStream)) {
			IOUtils.closeQuietly(currentStateHandleInStream);
		}
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@Override
public K next() {
	if (!hasNext()) {
		throw new NoSuchElementException("Failed to access state [" + state + "]");
	}
	try {
		byte[] key = iterator.key();
			DataInputViewStreamWrapper dataInput = new DataInputViewStreamWrapper(
			new ByteArrayInputStreamWithPos(key, keyGroupPrefixBytes, key.length - keyGroupPrefixBytes));
		K value = keySerializer.deserialize(dataInput);
		iterator.next();
		return value;
	} catch (IOException e) {
		throw new FlinkRuntimeException("Failed to access state [" + state + "]", e);
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@Override
public R get() throws IOException {
	try {
		// prepare the current key and namespace for RocksDB lookup
		writeCurrentKeyWithGroupAndNamespace();
		final byte[] key = keySerializationStream.toByteArray();

		// get the current value
		final byte[] valueBytes = backend.db.get(columnFamily, key);

		if (valueBytes == null) {
			return null;
		}

		ACC accumulator = valueSerializer.deserialize(new DataInputViewStreamWrapper(new ByteArrayInputStreamWithPos(valueBytes)));
		return aggFunction.getResult(accumulator);
	}
	catch (IOException | RocksDBException e) {
		throw new IOException("Error while retrieving value from RocksDB", e);
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
	in.defaultReadObject();

	int collectionLength = in.readInt();
	List<T> list = new ArrayList<T>(collectionLength);

	if (collectionLength > 0) {
		try {
			DataInputViewStreamWrapper wrapper = new DataInputViewStreamWrapper(in);
			for (int i = 0; i < collectionLength; i++){
				T element = serializer.deserialize(wrapper);
				list.add(element);
			}
		}
		catch (Throwable t) {
			throw new IOException("Error while deserializing element from collection", t);
		}
	}

	dataSet = list;
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {
	// read the non-transient fields
	in.defaultReadObject();

	// read the default value field
	boolean hasDefaultValue = in.readBoolean();
	if (hasDefaultValue) {
		int size = in.readInt();

		byte[] buffer = new byte[size];

		in.readFully(buffer);

		try (ByteArrayInputStream bais = new ByteArrayInputStream(buffer);
				DataInputViewStreamWrapper inView = new DataInputViewStreamWrapper(bais))
		{
			defaultValue = serializer.deserialize(inView);
		}
		catch (Exception e) {
			throw new IOException("Unable to deserialize default value.", e);
		}
	} else {
		defaultValue = null;
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
/**
 * Fill in the statistics. The last modification time and the total input size are prefilled.
 *
 * @param files
 *        The files that are associated with this block input format.
 * @param stats
 *        The pre-filled statistics.
 */
protected SequentialStatistics createStatistics(List<FileStatus> files, FileBaseStatistics stats)
		throws IOException {
	if (files.isEmpty()) {
		return null;
	}

	BlockInfo blockInfo = new BlockInfo();
	long totalCount = 0;
	for (FileStatus file : files) {
		// invalid file
		if (file.getLen() < blockInfo.getInfoSize()) {
			continue;
		}

		FileSystem fs = file.getPath().getFileSystem();
		try (FSDataInputStream fdis = fs.open(file.getPath(), blockInfo.getInfoSize())) {
			fdis.seek(file.getLen() - blockInfo.getInfoSize());

			blockInfo.read(new DataInputViewStreamWrapper(fdis));
			totalCount += blockInfo.getAccumulatedRecordCount();
		}
	}

	final float avgWidth = totalCount == 0 ? 0 : ((float) stats.getTotalInputSize() / totalCount);
	return new SequentialStatistics(stats.getLastModificationTime(), stats.getTotalInputSize(), avgWidth,
		totalCount);
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
@PublicEvolving
@Override
public void reopen(FileInputSplit split, Tuple2<Long, Long> state) throws IOException {
	Preconditions.checkNotNull(split, "reopen() cannot be called on a null split.");
	Preconditions.checkNotNull(state, "reopen() cannot be called with a null initial state.");

	try {
		this.open(split);
	} finally {
		this.blockInfo = this.createAndReadBlockInfo();

		long blockPos = state.f0;
		this.readRecords = state.f1;

		this.stream.seek(this.splitStart + blockPos);
		this.blockBasedInput = new BlockBasedInput(this.stream, (int) blockPos, this.splitLength);
		this.dataInputStream = new DataInputViewStreamWrapper(blockBasedInput);
	}
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies that reconfiguring with a config snapshot of a preceding POJO serializer
 * with different POJO type will result in INCOMPATIBLE.
 */
@Test
public void testReconfigureWithDifferentPojoType() throws Exception {
	PojoSerializer<SubTestUserClassB> pojoSerializer1 = (PojoSerializer<SubTestUserClassB>)
		TypeExtractor.getForClass(SubTestUserClassB.class).createSerializer(new ExecutionConfig());

	// snapshot configuration and serialize to bytes
	TypeSerializerConfigSnapshot pojoSerializerConfigSnapshot = pojoSerializer1.snapshotConfiguration();
	byte[] serializedConfig;
	try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
		TypeSerializerSerializationUtil.writeSerializerConfigSnapshot(new DataOutputViewStreamWrapper(out), pojoSerializerConfigSnapshot);
		serializedConfig = out.toByteArray();
	}

	PojoSerializer<SubTestUserClassA> pojoSerializer2 = (PojoSerializer<SubTestUserClassA>)
		TypeExtractor.getForClass(SubTestUserClassA.class).createSerializer(new ExecutionConfig());

	// read configuration again from bytes
	try(ByteArrayInputStream in = new ByteArrayInputStream(serializedConfig)) {
		pojoSerializerConfigSnapshot = TypeSerializerSerializationUtil.readSerializerConfigSnapshot(
			new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader());
	}

	CompatibilityResult<SubTestUserClassA> compatResult = pojoSerializer2.ensureCompatibility(pojoSerializerConfigSnapshot);
	assertTrue(compatResult.isRequiresMigration());
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies that reading and writing configuration snapshots work correctly.
 */
@Test
public void testSerializeConfigurationSnapshots() throws Exception {
	TestConfigSnapshot configSnapshot1 = new TestConfigSnapshot(1, "foo");
	TestConfigSnapshot configSnapshot2 = new TestConfigSnapshot(2, "bar");

	byte[] serializedConfig;
	try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
		TypeSerializerUtil.writeSerializerConfigSnapshots(
			new DataOutputViewStreamWrapper(out),
			configSnapshot1,
			configSnapshot2);

		serializedConfig = out.toByteArray();
	}

	TypeSerializerConfigSnapshot[] restoredConfigs;
	try (ByteArrayInputStream in = new ByteArrayInputStream(serializedConfig)) {
		restoredConfigs = TypeSerializerUtil.readSerializerConfigSnapshots(
			new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader());
	}

	assertEquals(2, restoredConfigs.length);
	assertEquals(configSnapshot1, restoredConfigs[0]);
	assertEquals(configSnapshot2, restoredConfigs[1]);
}
 

import org.apache.flink.core.memory.DataInputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies that deserializing config snapshots fail if the config class could not be found.
 */
@Test
public void testFailsWhenConfigurationSnapshotClassNotFound() throws Exception {
	byte[] serializedConfig;
	try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
		TypeSerializerUtil.writeSerializerConfigSnapshot(
			new DataOutputViewStreamWrapper(out), new TestConfigSnapshot(123, "foobar"));
		serializedConfig = out.toByteArray();
	}

	try (ByteArrayInputStream in = new ByteArrayInputStream(serializedConfig)) {
		// read using a dummy classloader
		TypeSerializerUtil.readSerializerConfigSnapshot(
			new DataInputViewStreamWrapper(in), new URLClassLoader(new URL[0], null));
		fail("Expected a ClassNotFoundException wrapped in IOException");
	} catch (IOException expected) {
		// test passes
	}
}