KSQL-12911 | Transform to Process Migration (2nd PR)

ksqldb-engine/src/main/java/io/confluent/ksql/execution/common/operators/ProjectOperator.java

@@ -22,7 +22,6 @@
 import io.confluent.ksql.execution.common.QueryRow;
 import io.confluent.ksql.execution.common.QueryRowImpl;
 import io.confluent.ksql.execution.plan.SelectExpression;
-import io.confluent.ksql.execution.streams.materialization.PullProcessingContext;
 import io.confluent.ksql.execution.transform.KsqlTransformer;
 import io.confluent.ksql.execution.transform.select.SelectValueMapper;
 import io.confluent.ksql.execution.transform.select.SelectValueMapperFactory;
@@ -107,9 +106,7 @@ public Object next() {
 
     final GenericRow mapped = transformer.transform(
         row.key(),
-        intermediate,
-        new PullProcessingContext(row.rowTime())
-    );
+        intermediate);
     validateProjection(mapped, logicalNode.getSchema());
 
     return QueryRowImpl.of(logicalNode.getSchema(),

ksqldb-engine/src/main/java/io/confluent/ksql/execution/common/operators/SelectOperator.java

@@ -21,7 +21,6 @@
 import io.confluent.ksql.execution.common.QueryRow;
 import io.confluent.ksql.execution.common.QueryRowImpl;
 import io.confluent.ksql.execution.streams.SqlPredicateFactory;
-import io.confluent.ksql.execution.streams.materialization.PullProcessingContext;
 import io.confluent.ksql.execution.transform.KsqlTransformer;
 import io.confluent.ksql.execution.transform.sqlpredicate.SqlPredicate;
 import io.confluent.ksql.logging.processing.ProcessingLogger;
@@ -84,10 +83,7 @@ public Object next() {
   private Optional<QueryRow> transformRow(final QueryRow queryRow) {
     final GenericRow intermediate = PhysicalOperatorUtil.getIntermediateRow(
         queryRow, logicalNode.getAddAdditionalColumnsToIntermediateSchema());
-    return transformer.transform(
-        queryRow.key(),
-        intermediate,
-        new PullProcessingContext(queryRow.rowTime()))
+    return transformer.transform(queryRow.key(), intermediate)
         .map(r -> QueryRowImpl.of(
             logicalNode.getIntermediateSchema(),
             queryRow.key(),

ksqldb-engine/src/test/java/io/confluent/ksql/execution/common/operators/ProjectOperatorTest.java

@@ -25,9 +25,6 @@
 import io.confluent.ksql.GenericKey;
 import io.confluent.ksql.GenericRow;
 import io.confluent.ksql.Window;
-import io.confluent.ksql.execution.common.operators.AbstractPhysicalOperator;
-import io.confluent.ksql.execution.common.operators.ProjectOperator;
-import io.confluent.ksql.execution.streams.materialization.PullProcessingContext;
 import io.confluent.ksql.execution.transform.KsqlTransformer;
 import io.confluent.ksql.execution.transform.select.SelectValueMapper;
 import io.confluent.ksql.execution.transform.select.SelectValueMapperFactory.SelectValueMapperFactorySupplier;
@@ -122,7 +119,7 @@ public void shouldProjectAllColumnsWhenSelectStarNonWindowed() {
     when(selectValueMapperFactorySupplier.create(any(), any()))
         .thenReturn(selectValueMapper);
     when(selectValueMapper.getTransformer(logger)).thenReturn(transformer);
-    when(transformer.transform(A_KEY, row.value(), new PullProcessingContext(12335L)))
+    when(transformer.transform(A_KEY, row.value()))
         .thenReturn(GenericRow.genericRow("k", "a", "b"));
     projectOperator.open();
 
@@ -156,7 +153,7 @@ public void shouldProjectAllColumnsWhenSelectStarWindowed() {
     when(selectValueMapperFactorySupplier.create(any(), any()))
         .thenReturn(selectValueMapper);
     when(selectValueMapper.getTransformer(logger)).thenReturn(transformer);
-    when(transformer.transform(A_KEY, windowedRow.value(), new PullProcessingContext(A_ROWTIME)))
+    when(transformer.transform(A_KEY, windowedRow.value()))
         .thenReturn(GenericRow.genericRow("k", A_WINDOW.start().toEpochMilli(), A_WINDOW.end().toEpochMilli(), "a", "b"));
     projectOperator.open();
 
@@ -194,15 +191,15 @@ public void shouldCallTransformWithCorrectArguments() {
     when(selectValueMapperFactorySupplier.create(any(), any()))
         .thenReturn(selectValueMapper);
     when(selectValueMapper.getTransformer(logger)).thenReturn(transformer);
-    when(transformer.transform(any(), any(), any())).thenReturn(GenericRow.genericRow("k", "a", "b"));
+    when(transformer.transform(any(), any())).thenReturn(GenericRow.genericRow("k", "a", "b"));
     projectOperator.open();
 
     // When:
     projectOperator.next();
 
     // Then:
     verify(transformer).transform(
-        A_KEY, GenericRow.genericRow("a", "b", 12335L, "k", 12335L, "k"), new PullProcessingContext(12335L));
+        A_KEY, GenericRow.genericRow("a", "b", 12335L, "k", 12335L, "k"));
   }
 
   @Test
@@ -227,7 +224,7 @@ public void shouldCallTransformWithCorrectArgumentsWindowed() {
     when(selectValueMapperFactorySupplier.create(any(), any()))
         .thenReturn(selectValueMapper);
     when(selectValueMapper.getTransformer(logger)).thenReturn(transformer);
-    when(transformer.transform(any(), any(), any())).thenReturn(GenericRow.genericRow("k", "a", "b"));
+    when(transformer.transform(any(), any())).thenReturn(GenericRow.genericRow("k", "a", "b"));
     projectOperator.open();
 
     // When:
@@ -237,8 +234,8 @@ public void shouldCallTransformWithCorrectArgumentsWindowed() {
     verify(transformer).transform(
         A_KEY,
         GenericRow.genericRow("a", "b", 12335L, "k", A_WINDOW.start().toEpochMilli(), A_WINDOW.end().toEpochMilli(),
-                              12335L, "k", A_WINDOW.start().toEpochMilli(), A_WINDOW.end().toEpochMilli()),
-        new PullProcessingContext(12335L));
+                              12335L, "k", A_WINDOW.start().toEpochMilli(), A_WINDOW.end().toEpochMilli())
+    );
   }
 
   @Test
@@ -266,7 +263,7 @@ public void shouldProjectOnlyKeyNonWindowed() {
     when(selectValueMapperFactorySupplier.create(any(), any()))
         .thenReturn(selectValueMapper);
     when(selectValueMapper.getTransformer(logger)).thenReturn(transformer);
-    when(transformer.transform(A_KEY, row.value(), new PullProcessingContext(12335L)))
+    when(transformer.transform(A_KEY, row.value()))
              .thenReturn(GenericRow.genericRow("k"));
     projectOperator.open();
 
@@ -302,7 +299,7 @@ public void shouldProjectOnlyValueNonWindowed() {
     when(selectValueMapperFactorySupplier.create(any(), any()))
         .thenReturn(selectValueMapper);
     when(selectValueMapper.getTransformer(logger)).thenReturn(transformer);
-    when(transformer.transform(A_KEY, row.value(), new PullProcessingContext(12335L)))
+    when(transformer.transform(A_KEY, row.value()))
         .thenReturn(GenericRow.genericRow("b"));
     projectOperator.open();
 
@@ -338,7 +335,7 @@ public void shouldProjectOnlyWindowStartWindowed() {
     when(selectValueMapperFactorySupplier.create(any(), any()))
         .thenReturn(selectValueMapper);
     when(selectValueMapper.getTransformer(logger)).thenReturn(transformer);
-    when(transformer.transform(A_KEY, windowedRow.value(), new PullProcessingContext(12335L)))
+    when(transformer.transform(A_KEY, windowedRow.value()))
         .thenReturn(GenericRow.genericRow(A_WINDOW.start().toEpochMilli()));
     projectOperator.open();
 
@@ -376,7 +373,7 @@ public void shouldProjectKeyAndValueNonWindowed() {
     when(selectValueMapperFactorySupplier.create(any(), any()))
         .thenReturn(selectValueMapper);
     when(selectValueMapper.getTransformer(logger)).thenReturn(transformer);
-    when(transformer.transform(A_KEY, row.value(), new PullProcessingContext(12335L)))
+    when(transformer.transform(A_KEY, row.value()))
         .thenReturn(GenericRow.genericRow("k","b"));
     projectOperator.open();
 
@@ -417,8 +414,8 @@ public void shouldProjectKeyAndValueWindowed() {
     when(selectValueMapper.getTransformer(logger)).thenReturn(transformer);
     when(transformer.transform(
         A_KEY,
-        windowedRow.value(),
-        new PullProcessingContext(12335L)))
+        windowedRow.value()
+    ))
         .thenReturn(GenericRow.genericRow("k", "b"));
     projectOperator.open();
 

ksqldb-engine/src/test/java/io/confluent/ksql/execution/common/operators/SelectOperatorTest.java

@@ -22,10 +22,7 @@
 
 import io.confluent.ksql.GenericKey;
 import io.confluent.ksql.GenericRow;
-import io.confluent.ksql.execution.common.operators.AbstractPhysicalOperator;
-import io.confluent.ksql.execution.common.operators.SelectOperator;
 import io.confluent.ksql.execution.streams.SqlPredicateFactory;
-import io.confluent.ksql.execution.streams.materialization.PullProcessingContext;
 import io.confluent.ksql.Window;
 import io.confluent.ksql.execution.transform.KsqlTransformer;
 import io.confluent.ksql.execution.transform.sqlpredicate.SqlPredicate;
@@ -120,7 +117,7 @@ public void shouldSelectKeyNonWindowed() {
         GenericRow.genericRow("a", "b", A_ROWTIME, "k"),
         A_ROWTIME
     );
-    when(transformer.transform(A_KEY, intermediateRow.value(), new PullProcessingContext(12335L)))
+    when(transformer.transform(A_KEY, intermediateRow.value()))
         .thenReturn(Optional.of(GenericRow.genericRow("a", "b", A_ROWTIME, "k")));
     selectOperator.open();
 
@@ -158,7 +155,7 @@ public void shouldSelectKeyWindowed() {
         GenericRow.genericRow("a", "b", A_ROWTIME, "k", A_WINDOW.start().toEpochMilli(), A_WINDOW.end().toEpochMilli()),
         A_ROWTIME
     );
-    when(transformer.transform(A_KEY, intermediateWindowedRow.value(), new PullProcessingContext(12335L)))
+    when(transformer.transform(A_KEY, intermediateWindowedRow.value()))
         .thenReturn(Optional.of(GenericRow.genericRow("a", "b", A_ROWTIME, "k", A_WINDOW.start().toEpochMilli(), A_WINDOW.end().toEpochMilli())));
     selectOperator.open();
 

ksqldb-engine/src/test/java/io/confluent/ksql/integration/SelectValueMapperIntegrationTest.java

@@ -71,8 +71,7 @@ public void shouldSelectChosenColumns() {
     // When:
     final GenericRow transformed = selectTransformer.transform(
         NON_WINDOWED_KEY,
-        genericRow("hi", "bye", 2.0D, "blah", "dar", ImmutableList.of(), 1521834663L, 0, 0L, 1L),
-        ctx
+        genericRow("hi", "bye", 2.0D, "blah", "dar", ImmutableList.of(), 1521834663L, 0, 0L, 1L)
     );
 
     // Then:
@@ -90,8 +89,7 @@ public void shouldApplyUdfsToColumns() {
     // When:
     final GenericRow row = selectTransformer.transform(
         NON_WINDOWED_KEY,
-        genericRow("foo", "whatever", 6.9D, "boo", "hoo", 0, 0L, ImmutableList.of(), 1521834663L, 2L),
-        ctx
+        genericRow("foo", "whatever", 6.9D, "boo", "hoo", 0, 0L, ImmutableList.of(), 1521834663L, 2L)
     );
 
     // Then: