feat: Adds Scalable Push Query physical operators (#7430)

* feat: Adds Scalable Push Query physical operators
confluentinc · May 13, 2021 · 100767d · 100767d
1 parent 7d0fec6
commit 100767d
Show file tree

Hide file tree

Showing 28 changed files with 988 additions and 100 deletions.
diff --git a/...l/operators/AbstractPhysicalOperator.java → ...n/operators/AbstractPhysicalOperator.java b/...l/operators/AbstractPhysicalOperator.java → ...n/operators/AbstractPhysicalOperator.java
@@ -13,7 +13,7 @@
  * specific language governing permissions and limitations under the License.
  */
 
-package io.confluent.ksql.physical.pull.operators;
+package io.confluent.ksql.physical.common.operators;
 
 import io.confluent.ksql.planner.plan.PlanNode;
 import java.util.List;

diff --git a/...l/operators/PullPhysicalOperatorUtil.java → ...ommon/operators/PhysicalOperatorUtil.java b/...l/operators/PullPhysicalOperatorUtil.java → ...ommon/operators/PhysicalOperatorUtil.java
@@ -13,16 +13,16 @@
  * specific language governing permissions and limitations under the License.
  */
 
-package io.confluent.ksql.physical.pull.operators;
+package io.confluent.ksql.physical.common.operators;
 
 import io.confluent.ksql.GenericKey;
 import io.confluent.ksql.GenericRow;
 import io.confluent.ksql.execution.streams.materialization.TableRow;
 import java.util.List;
 
-final class PullPhysicalOperatorUtil {
+final class PhysicalOperatorUtil {
 
-  private PullPhysicalOperatorUtil() {
+  private PhysicalOperatorUtil() {
 
   }
 

diff --git a/...sical/pull/operators/ProjectOperator.java → ...cal/common/operators/ProjectOperator.java b/...sical/pull/operators/ProjectOperator.java → ...cal/common/operators/ProjectOperator.java
@@ -13,7 +13,7 @@
  * specific language governing permissions and limitations under the License.
  */
 
-package io.confluent.ksql.physical.pull.operators;
+package io.confluent.ksql.physical.common.operators;
 
 import com.google.common.annotations.VisibleForTesting;
 import io.confluent.ksql.GenericRow;
@@ -26,7 +26,7 @@
 import io.confluent.ksql.execution.transform.select.SelectValueMapperFactory.SelectValueMapperFactorySupplier;
 import io.confluent.ksql.logging.processing.ProcessingLogger;
 import io.confluent.ksql.planner.plan.PlanNode;
-import io.confluent.ksql.planner.plan.PullProjectNode;
+import io.confluent.ksql.planner.plan.QueryProjectNode;
 import io.confluent.ksql.schema.ksql.Column;
 import io.confluent.ksql.schema.ksql.LogicalSchema;
 import java.util.ArrayList;
@@ -38,15 +38,15 @@ public class ProjectOperator extends AbstractPhysicalOperator implements UnaryPh
 
   private final ProcessingLogger logger;
   private final SelectValueMapperFactorySupplier selectValueMapperFactorySupplier;
-  private final PullProjectNode logicalNode;
+  private final QueryProjectNode logicalNode;
 
   private AbstractPhysicalOperator child;
   private TableRow row;
   private KsqlTransformer<Object, GenericRow> transformer;
 
   public ProjectOperator(
       final ProcessingLogger logger,
-      final PullProjectNode logicalNode
+      final QueryProjectNode logicalNode
   ) {
     this(
         logger,
@@ -58,7 +58,7 @@ public ProjectOperator(
   @VisibleForTesting
   ProjectOperator(
       final ProcessingLogger logger,
-      final PullProjectNode logicalNode,
+      final QueryProjectNode logicalNode,
       final SelectValueMapperFactorySupplier selectValueMapperFactorySupplier
   ) {
     this.logger = Objects.requireNonNull(logger, "logger");
@@ -89,7 +89,7 @@ public Object next() {
       return null;
     }
 
-    final GenericRow intermediate = PullPhysicalOperatorUtil.getIntermediateRow(
+    final GenericRow intermediate = PhysicalOperatorUtil.getIntermediateRow(
         row, logicalNode.getAddAdditionalColumnsToIntermediateSchema());
 
     if (logicalNode.getIsSelectStar()) {

diff --git a/...ysical/pull/operators/SelectOperator.java → ...ical/common/operators/SelectOperator.java b/...ysical/pull/operators/SelectOperator.java → ...ical/common/operators/SelectOperator.java
@@ -13,7 +13,7 @@
  * specific language governing permissions and limitations under the License.
  */
 
-package io.confluent.ksql.physical.pull.operators;
+package io.confluent.ksql.physical.common.operators;
 
 import com.google.common.annotations.VisibleForTesting;
 import io.confluent.ksql.GenericRow;
@@ -26,28 +26,28 @@
 import io.confluent.ksql.execution.transform.sqlpredicate.SqlPredicate;
 import io.confluent.ksql.logging.processing.ProcessingLogger;
 import io.confluent.ksql.planner.plan.PlanNode;
-import io.confluent.ksql.planner.plan.PullFilterNode;
+import io.confluent.ksql.planner.plan.QueryFilterNode;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 
 public class SelectOperator extends AbstractPhysicalOperator implements UnaryPhysicalOperator {
 
-  private final PullFilterNode logicalNode;
+  private final QueryFilterNode logicalNode;
   private final ProcessingLogger logger;
   private final SqlPredicate predicate;
 
   private AbstractPhysicalOperator child;
   private KsqlTransformer<Object, Optional<GenericRow>> transformer;
   private TableRow row;
 
-  public SelectOperator(final PullFilterNode logicalNode, final ProcessingLogger logger) {
+  public SelectOperator(final QueryFilterNode logicalNode, final ProcessingLogger logger) {
     this(logicalNode, logger, SqlPredicate::new);
   }
 
   @VisibleForTesting
   SelectOperator(
-      final PullFilterNode logicalNode,
+      final QueryFilterNode logicalNode,
       final ProcessingLogger logger,
       final SqlPredicateFactory predicateFactory
   ) {
@@ -80,7 +80,7 @@ public Object next() {
   }
 
   private Optional<TableRow> transformRow(final TableRow tableRow) {
-    final GenericRow intermediate = PullPhysicalOperatorUtil.getIntermediateRow(
+    final GenericRow intermediate = PhysicalOperatorUtil.getIntermediateRow(
         tableRow, logicalNode.getAddAdditionalColumnsToIntermediateSchema());
     return transformer.transform(
         tableRow.key(),

diff --git a/...pull/operators/UnaryPhysicalOperator.java → ...mmon/operators/UnaryPhysicalOperator.java b/...pull/operators/UnaryPhysicalOperator.java → ...mmon/operators/UnaryPhysicalOperator.java
@@ -13,7 +13,7 @@
  * specific language governing permissions and limitations under the License.
  */
 
-package io.confluent.ksql.physical.pull.operators;
+package io.confluent.ksql.physical.common.operators;
 
 /**
  * Represents a physical operator of the physical plan that has a single child.

diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/physical/pull/PullPhysicalPlan.java b/ksqldb-engine/src/main/java/io/confluent/ksql/physical/pull/PullPhysicalPlan.java
@@ -19,7 +19,7 @@
 import io.confluent.ksql.execution.streams.materialization.Locator.KsqlKey;
 import io.confluent.ksql.execution.streams.materialization.Locator.KsqlPartitionLocation;
 import io.confluent.ksql.execution.streams.materialization.Materialization;
-import io.confluent.ksql.physical.pull.operators.AbstractPhysicalOperator;
+import io.confluent.ksql.physical.common.operators.AbstractPhysicalOperator;
 import io.confluent.ksql.physical.pull.operators.DataSourceOperator;
 import io.confluent.ksql.planner.plan.KeyConstraint;
 import io.confluent.ksql.planner.plan.KeyConstraint.ConstraintOperator;

diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/physical/pull/PullPhysicalPlanBuilder.java b/ksqldb-engine/src/main/java/io/confluent/ksql/physical/pull/PullPhysicalPlanBuilder.java
@@ -25,14 +25,14 @@
 import io.confluent.ksql.logging.processing.ProcessingLogger;
 import io.confluent.ksql.metastore.model.DataSource;
 import io.confluent.ksql.name.SourceName;
+import io.confluent.ksql.physical.common.operators.AbstractPhysicalOperator;
+import io.confluent.ksql.physical.common.operators.ProjectOperator;
+import io.confluent.ksql.physical.common.operators.SelectOperator;
 import io.confluent.ksql.physical.pull.PullPhysicalPlan.PullPhysicalPlanType;
 import io.confluent.ksql.physical.pull.PullPhysicalPlan.PullSourceType;
-import io.confluent.ksql.physical.pull.operators.AbstractPhysicalOperator;
 import io.confluent.ksql.physical.pull.operators.DataSourceOperator;
 import io.confluent.ksql.physical.pull.operators.KeyedTableLookupOperator;
 import io.confluent.ksql.physical.pull.operators.KeyedWindowedTableLookupOperator;
-import io.confluent.ksql.physical.pull.operators.ProjectOperator;
-import io.confluent.ksql.physical.pull.operators.SelectOperator;
 import io.confluent.ksql.physical.pull.operators.TableScanOperator;
 import io.confluent.ksql.physical.pull.operators.WindowedTableScanOperator;
 import io.confluent.ksql.planner.LogicalPlanNode;
@@ -42,8 +42,8 @@
 import io.confluent.ksql.planner.plan.NonKeyConstraint;
 import io.confluent.ksql.planner.plan.OutputNode;
 import io.confluent.ksql.planner.plan.PlanNode;
-import io.confluent.ksql.planner.plan.PullFilterNode;
-import io.confluent.ksql.planner.plan.PullProjectNode;
+import io.confluent.ksql.planner.plan.QueryFilterNode;
+import io.confluent.ksql.planner.plan.QueryProjectNode;
 import io.confluent.ksql.query.QueryId;
 import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.util.PersistentQueryMetadata;
@@ -109,10 +109,10 @@ public PullPhysicalPlan buildPullPhysicalPlan(final LogicalPlanNode logicalPlanN
     AbstractPhysicalOperator rootPhysicalOp = null;
     while (true) {
       AbstractPhysicalOperator currentPhysicalOp = null;
-      if (currentLogicalNode instanceof PullProjectNode) {
-        currentPhysicalOp = translateProjectNode((PullProjectNode)currentLogicalNode);
-      } else if (currentLogicalNode instanceof PullFilterNode) {
-        currentPhysicalOp = translateFilterNode((PullFilterNode) currentLogicalNode);
+      if (currentLogicalNode instanceof QueryProjectNode) {
+        currentPhysicalOp = translateProjectNode((QueryProjectNode)currentLogicalNode);
+      } else if (currentLogicalNode instanceof QueryFilterNode) {
+        currentPhysicalOp = translateFilterNode((QueryFilterNode) currentLogicalNode);
         seenSelectOperator = true;
       } else if (currentLogicalNode instanceof DataSourceNode) {
         currentPhysicalOp = translateDataSourceNode(
@@ -154,7 +154,7 @@ public PullPhysicalPlan buildPullPhysicalPlan(final LogicalPlanNode logicalPlanN
         dataSourceOperator);
   }
 
-  private ProjectOperator translateProjectNode(final PullProjectNode logicalNode) {
+  private ProjectOperator translateProjectNode(final QueryProjectNode logicalNode) {
     final ProcessingLogger logger = processingLogContext
         .getLoggerFactory()
         .getLogger(
@@ -168,7 +168,7 @@ private ProjectOperator translateProjectNode(final PullProjectNode logicalNode)
     );
   }
 
-  private SelectOperator translateFilterNode(final PullFilterNode logicalNode) {
+  private SelectOperator translateFilterNode(final QueryFilterNode logicalNode) {
     lookupConstraints = logicalNode.getLookupConstraints();
 
     final ProcessingLogger logger = processingLogContext

diff --git a/...ine/src/main/java/io/confluent/ksql/physical/pull/operators/KeyedTableLookupOperator.java b/...ine/src/main/java/io/confluent/ksql/physical/pull/operators/KeyedTableLookupOperator.java
@@ -21,6 +21,8 @@
 import io.confluent.ksql.execution.streams.materialization.Locator.KsqlPartitionLocation;
 import io.confluent.ksql.execution.streams.materialization.Materialization;
 import io.confluent.ksql.execution.streams.materialization.Row;
+import io.confluent.ksql.physical.common.operators.AbstractPhysicalOperator;
+import io.confluent.ksql.physical.common.operators.UnaryPhysicalOperator;
 import io.confluent.ksql.planner.plan.DataSourceNode;
 import io.confluent.ksql.planner.plan.PlanNode;
 import java.util.Iterator;

diff --git a/...main/java/io/confluent/ksql/physical/pull/operators/KeyedWindowedTableLookupOperator.java b/...main/java/io/confluent/ksql/physical/pull/operators/KeyedWindowedTableLookupOperator.java
@@ -19,10 +19,12 @@
 import io.confluent.ksql.execution.streams.materialization.Locator.KsqlPartitionLocation;
 import io.confluent.ksql.execution.streams.materialization.Materialization;
 import io.confluent.ksql.execution.streams.materialization.WindowedRow;
+import io.confluent.ksql.physical.common.operators.AbstractPhysicalOperator;
+import io.confluent.ksql.physical.common.operators.UnaryPhysicalOperator;
 import io.confluent.ksql.planner.plan.DataSourceNode;
 import io.confluent.ksql.planner.plan.KeyConstraint.KeyConstraintKey;
 import io.confluent.ksql.planner.plan.PlanNode;
-import io.confluent.ksql.planner.plan.PullFilterNode.WindowBounds;
+import io.confluent.ksql.planner.plan.QueryFilterNode.WindowBounds;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;

diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/physical/pull/operators/TableScanOperator.java b/ksqldb-engine/src/main/java/io/confluent/ksql/physical/pull/operators/TableScanOperator.java
@@ -18,6 +18,8 @@
 import io.confluent.ksql.execution.streams.materialization.Locator.KsqlPartitionLocation;
 import io.confluent.ksql.execution.streams.materialization.Materialization;
 import io.confluent.ksql.execution.streams.materialization.Row;
+import io.confluent.ksql.physical.common.operators.AbstractPhysicalOperator;
+import io.confluent.ksql.physical.common.operators.UnaryPhysicalOperator;
 import io.confluent.ksql.planner.plan.DataSourceNode;
 import io.confluent.ksql.planner.plan.PlanNode;
 import java.util.Iterator;

diff --git a/...ne/src/main/java/io/confluent/ksql/physical/pull/operators/WindowedTableScanOperator.java b/...ne/src/main/java/io/confluent/ksql/physical/pull/operators/WindowedTableScanOperator.java
@@ -19,6 +19,8 @@
 import io.confluent.ksql.execution.streams.materialization.Locator.KsqlPartitionLocation;
 import io.confluent.ksql.execution.streams.materialization.Materialization;
 import io.confluent.ksql.execution.streams.materialization.WindowedRow;
+import io.confluent.ksql.physical.common.operators.AbstractPhysicalOperator;
+import io.confluent.ksql.physical.common.operators.UnaryPhysicalOperator;
 import io.confluent.ksql.planner.plan.DataSourceNode;
 import io.confluent.ksql.planner.plan.PlanNode;
 import java.util.Iterator;