@Test public void testSnapshot() throws UnknownHostException { final RowTypeInfo type = new RowTypeInfo(Types.STRING, Types.LONG); TestMaterializedCollectStreamResult<?> result = null; try { result = new TestMaterializedCollectStreamResult<>( type, new ExecutionConfig(), InetAddress.getLocalHost(), 0, Integer.MAX_VALUE); result.isRetrieving = true; result.processRecord(Tuple2.of(true, Row.of("A", 1))); result.processRecord(Tuple2.of(true, Row.of("B", 1))); result.processRecord(Tuple2.of(true, Row.of("A", 1))); result.processRecord(Tuple2.of(true, Row.of("C", 2))); assertEquals(TypedResult.payload(4), result.snapshot(1)); assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(1)); assertEquals(Collections.singletonList(Row.of("B", 1)), result.retrievePage(2)); assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(3)); assertEquals(Collections.singletonList(Row.of("C", 2)), result.retrievePage(4)); result.processRecord(Tuple2.of(false, Row.of("A", 1))); assertEquals(TypedResult.payload(3), result.snapshot(1)); assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(1)); assertEquals(Collections.singletonList(Row.of("B", 1)), result.retrievePage(2)); assertEquals(Collections.singletonList(Row.of("C", 2)), result.retrievePage(3)); result.processRecord(Tuple2.of(false, Row.of("C", 2))); result.processRecord(Tuple2.of(false, Row.of("A", 1))); assertEquals(TypedResult.payload(1), result.snapshot(1)); assertEquals(Collections.singletonList(Row.of("B", 1)), result.retrievePage(1)); } finally { if (result != null) { result.close(); } } }
@Override public TypedResult<Integer> snapshot(int pageSize) { if (pageSize < 1) { throw new SqlExecutionException("Page size must be greater than 0."); } synchronized (resultLock) { if ((!isRetrieving() && isLastSnapshot) || executionException != null) { return handleMissingResult(); } else if (!isRetrieving()) { isLastSnapshot = true; } this.pageSize = pageSize; snapshot.clear(); for (int i = validRowPosition; i < materializedTable.size(); i++) { snapshot.add(materializedTable.get(i)); } pageCount = Math.max(1, (int) Math.ceil(((double) snapshot.size() / pageSize))); return TypedResult.payload(pageCount); } }
MaterializedCollectStreamResult extends CollectStreamResult<C> implements MaterializedResult<C> { @Override public TypedResult<Integer> snapshot(int pageSize) { if (pageSize < 1) { throw new SqlExecutionException("Page size must be greater than 0."); } synchronized (resultLock) { if ((!isRetrieving() && isLastSnapshot) || executionException != null) { return handleMissingResult(); } else if (!isRetrieving()) { isLastSnapshot = true; } this.pageSize = pageSize; snapshot.clear(); for (int i = validRowPosition; i < materializedTable.size(); i++) { snapshot.add(materializedTable.get(i)); } pageCount = Math.max(1, (int) Math.ceil(((double) snapshot.size() / pageSize))); return TypedResult.payload(pageCount); } } }
MaterializedCollectStreamResult extends CollectStreamResult<C> implements MaterializedResult<C> { @Override public TypedResult<Integer> snapshot(int pageSize) { if (pageSize < 1) { throw new SqlExecutionException("Page size must be greater than 0."); } synchronized (resultLock) { if ((!isRetrieving() && isLastSnapshot) || executionException != null) { return handleMissingResult(); } else if (!isRetrieving()) { isLastSnapshot = true; } this.pageSize = pageSize; snapshot.clear(); for (int i = validRowPosition; i < materializedTable.size(); i++) { snapshot.add(materializedTable.get(i)); } pageCount = Math.max(1, (int) Math.ceil(((double) snapshot.size() / pageSize))); return TypedResult.payload(pageCount); } } @VisibleForTesting  MaterializedCollectStreamResult(
			RowTypeInfo outputType,
			ExecutionConfig config,
			InetAddress gatewayAddress,
			int gatewayPort,
			int maxRowCount,
			int overcommitThreshold);  MaterializedCollectStreamResult(
			RowTypeInfo outputType,
			ExecutionConfig config,
			InetAddress gatewayAddress,
			int gatewayPort,
			int maxRowCount); }
MaterializedCollectStreamResult extends CollectStreamResult<C> implements MaterializedResult<C> { @Override public TypedResult<Integer> snapshot(int pageSize) { if (pageSize < 1) { throw new SqlExecutionException("Page size must be greater than 0."); } synchronized (resultLock) { if ((!isRetrieving() && isLastSnapshot) || executionException != null) { return handleMissingResult(); } else if (!isRetrieving()) { isLastSnapshot = true; } this.pageSize = pageSize; snapshot.clear(); for (int i = validRowPosition; i < materializedTable.size(); i++) { snapshot.add(materializedTable.get(i)); } pageCount = Math.max(1, (int) Math.ceil(((double) snapshot.size() / pageSize))); return TypedResult.payload(pageCount); } } @VisibleForTesting  MaterializedCollectStreamResult(
			RowTypeInfo outputType,
			ExecutionConfig config,
			InetAddress gatewayAddress,
			int gatewayPort,
			int maxRowCount,
			int overcommitThreshold);  MaterializedCollectStreamResult(
			RowTypeInfo outputType,
			ExecutionConfig config,
			InetAddress gatewayAddress,
			int gatewayPort,
			int maxRowCount); @Override boolean isMaterialized(); @Override TypedResult<Integer> snapshot(int pageSize); @Override List<Row> retrievePage(int page); }
MaterializedCollectStreamResult extends CollectStreamResult<C> implements MaterializedResult<C> { @Override public TypedResult<Integer> snapshot(int pageSize) { if (pageSize < 1) { throw new SqlExecutionException("Page size must be greater than 0."); } synchronized (resultLock) { if ((!isRetrieving() && isLastSnapshot) || executionException != null) { return handleMissingResult(); } else if (!isRetrieving()) { isLastSnapshot = true; } this.pageSize = pageSize; snapshot.clear(); for (int i = validRowPosition; i < materializedTable.size(); i++) { snapshot.add(materializedTable.get(i)); } pageCount = Math.max(1, (int) Math.ceil(((double) snapshot.size() / pageSize))); return TypedResult.payload(pageCount); } } @VisibleForTesting  MaterializedCollectStreamResult(
			RowTypeInfo outputType,
			ExecutionConfig config,
			InetAddress gatewayAddress,
			int gatewayPort,
			int maxRowCount,
			int overcommitThreshold);  MaterializedCollectStreamResult(
			RowTypeInfo outputType,
			ExecutionConfig config,
			InetAddress gatewayAddress,
			int gatewayPort,
			int maxRowCount); @Override boolean isMaterialized(); @Override TypedResult<Integer> snapshot(int pageSize); @Override List<Row> retrievePage(int page); static final int MATERIALIZED_TABLE_MAX_INITIAL_CAPACITY; static final int MATERIALIZED_TABLE_MAX_OVERCOMMIT; static final double MATERIALIZED_TABLE_CAPACITY_FACTOR; static final double MATERIALIZED_TABLE_OVERCOMMIT_FACTOR; }