TransformerBase extends PipelineStageBase<T> implements Transformer<T>, HasLazyPrintTransformInfo<T> { @Override public Table transform(TableEnvironment tEnv, Table input) { Preconditions.checkArgument(input != null, "Input CAN NOT BE null!"); Preconditions.checkArgument( tableEnvOf(input) == tEnv, "The input table is not in the specified table environment."); return transform(input); } TransformerBase();  TransformerBase(Params params); @Override Table transform(TableEnvironment tEnv, Table input); Table transform(Table input); abstract BatchOperator transform(BatchOperator input); abstract StreamOperator transform(StreamOperator input);  }
@Test public void testFitBatchTable() { Long id = MLEnvironmentFactory.getNewMLEnvironmentId(); MLEnvironment env = MLEnvironmentFactory.get(id); DataSet<Integer> input = env.getExecutionEnvironment().fromElements(1, 2, 3); Table table = env.getBatchTableEnvironment().fromDataSet(input); FakeTransFormer transFormer = new FakeTransFormer(); transFormer.setMLEnvironmentId(id); transFormer.transform(env.getBatchTableEnvironment(), table); Assert.assertTrue(transFormer.batchTransformed); Assert.assertFalse(transFormer.streamTransformed); }
@Test public void testFitStreamTable() { Long id = MLEnvironmentFactory.getNewMLEnvironmentId(); MLEnvironment env = MLEnvironmentFactory.get(id); DataStream<Integer> input = env.getStreamExecutionEnvironment().fromElements(1, 2, 3); Table table = env.getStreamTableEnvironment().fromDataStream(input); FakeTransFormer transFormer = new FakeTransFormer(); transFormer.setMLEnvironmentId(id); transFormer.transform(env.getStreamTableEnvironment(), table); Assert.assertFalse(transFormer.batchTransformed); Assert.assertTrue(transFormer.streamTransformed); }