Removed TableFunctionBase and child classes in favor of QueryTableFun…

…ction. Removed VirtualRelationalTable and replaced with new classes. Updated the logical planner to keep structured columns.
DataSQRL · Oct 13, 2023 · b8031b7 · b8031b7
1 parent 1b08660
commit b8031b7
Show file tree

Hide file tree

Showing 36 changed files with 550 additions and 973 deletions.
diff --git a/sqrl-planner/sqrl-common/src/main/java/com/datasqrl/calcite/SqrlFramework.java b/sqrl-planner/sqrl-common/src/main/java/com/datasqrl/calcite/SqrlFramework.java
@@ -1,7 +1,11 @@
 package com.datasqrl.calcite;
 
 import com.datasqrl.calcite.type.TypeFactory;
+import com.datasqrl.canonicalizer.Name;
 import com.datasqrl.canonicalizer.NameCanonicalizer;
+
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.Getter;
 import org.apache.calcite.config.CalciteConnectionConfigImpl;
@@ -26,6 +30,7 @@ public class SqrlFramework {
   private AtomicInteger uniqueMacroInt = new AtomicInteger(0);
   private AtomicInteger uniqueTableInt = new AtomicInteger(0);
   private AtomicInteger uniqueColumnInt = new AtomicInteger(0);
+  private Map<Name,AtomicInteger> tableNameShadowing = new HashMap<Name,AtomicInteger>();
 
   public SqrlFramework() {
     this(null, HintStrategyTable.builder().build(), NameCanonicalizer.SYSTEM);
@@ -55,4 +60,14 @@ public QueryPlanner resetPlanner() {
     this.queryPlanner = new QueryPlanner(this);
     return this.queryPlanner;
   }
+
+  public int getUniqueTableId(Name tableName) {
+//    AtomicInteger counter = tableNameShadowing.get(tableName);
+//    if (counter==null) {
+//      counter = new AtomicInteger(0);
+//      tableNameShadowing.put(tableName,counter);
+//    }
+//    return counter.incrementAndGet();
+    return uniqueTableInt.incrementAndGet();
+  }
 }
diff --git a/sqrl-planner/sqrl-common/src/main/java/com/datasqrl/util/CalciteUtil.java b/sqrl-planner/sqrl-common/src/main/java/com/datasqrl/util/CalciteUtil.java
@@ -72,6 +72,18 @@ public static boolean hasNesting(RelDataType type) {
     return type.getFieldList().stream().map(t -> t.getType()).anyMatch(CalciteUtil::isNestedTable);
   }
 
+  public static RelBuilder projectOutNested(RelBuilder relBuilder) {
+    List<RelDataTypeField> fields = relBuilder.peek().getRowType().getFieldList();
+    List<RexNode> projects = new ArrayList<>(fields.size());
+    for (int i = 0; i < fields.size(); i++) {
+      if (!CalciteUtil.isNestedTable(fields.get(i).getType())) {
+        projects.add(relBuilder.field(i));
+      }
+    }
+    relBuilder.project(projects);
+    return relBuilder;
+  }
+
   public static boolean isArray(RelDataType type) {
     return type instanceof ArraySqlType;
   }

diff --git a/...lanner/sqrl-planner-local/src/main/java/com/datasqrl/graphql/APIConnectorManagerImpl.java b/...lanner/sqrl-planner-local/src/main/java/com/datasqrl/graphql/APIConnectorManagerImpl.java
@@ -19,10 +19,7 @@
 import com.datasqrl.plan.queries.APIQuery;
 import com.datasqrl.plan.queries.APISource;
 import com.datasqrl.plan.queries.APISubscription;
-import com.datasqrl.plan.table.CalciteTableFactory;
-import com.datasqrl.plan.table.RelDataType2UTBConverter;
-import com.datasqrl.plan.table.TableType;
-import com.datasqrl.plan.table.VirtualRelationalTable;
+import com.datasqrl.plan.table.*;
 import com.datasqrl.schema.SQRLTable;
 import com.datasqrl.schema.UniversalTable;
 import com.google.inject.Inject;
@@ -107,18 +104,18 @@ public TableSink getMutationSource(APISource source, Name mutationName) {
   @Override
   public TableSource addSubscription(APISubscription subscription, SQRLTable sqrlTable) {
     errors.checkFatal(logEngine.isPresent(), "Cannot create subscriptions because no log engine is configured");
-    errors.checkFatal(((VirtualRelationalTable) sqrlTable.getVt()).getRoot().getBase().getType()== TableType.STREAM,
+    errors.checkFatal(((ScriptRelationalTable) sqrlTable.getVt()).getRoot().getType()== TableType.STREAM,
         "Table %s for subscription %s is not a stream table", sqrlTable, subscription);
     //Check if we already exported it
     TableSource subscriptionSource;
     if (exports.containsKey(sqrlTable)) {
       subscriptionSource = exports.get(sqrlTable).getSource();
     } else {
       //otherwise create new log for it
-      String logId = ((VirtualRelationalTable) sqrlTable.getVt()).getNameId();
+      String logId = ((ScriptRelationalTable) sqrlTable.getVt()).getNameId();
       RelDataType2UTBConverter converter = new RelDataType2UTBConverter(typeFactory, 0,
           NameCanonicalizer.SYSTEM);
-      UniversalTable schema = converter.convert(sqrlTable.getPath(), ((VirtualRelationalTable) sqrlTable.getVt()).getRowType(),
+      UniversalTable schema = converter.convert(sqrlTable.getPath(), ((ScriptRelationalTable) sqrlTable.getVt()).getRowType(),
           null);
       Log log = logEngine.get().createLog(logId, schema);
       exports.put(sqrlTable, log);

diff --git a/sqrl-planner/sqrl-planner-local/src/main/java/com/datasqrl/plan/global/DAGAssembler.java b/sqrl-planner/sqrl-planner-local/src/main/java/com/datasqrl/plan/global/DAGAssembler.java
@@ -15,13 +15,13 @@
 import com.datasqrl.io.tables.TableSink;
 import com.datasqrl.canonicalizer.Name;
 import com.datasqrl.plan.RelStageRunner;
-import com.datasqrl.plan.local.generate.TableFunctionBase;
+import com.datasqrl.plan.local.generate.QueryTableFunction;
 import com.datasqrl.plan.rules.AnnotatedLP;
 import com.datasqrl.plan.rules.SQRLConverter;
 import com.datasqrl.plan.table.AbstractRelationalTable;
+import com.datasqrl.plan.table.LogicalNestedTable;
 import com.datasqrl.plan.table.PhysicalRelationalTable;
-import com.datasqrl.plan.table.ScriptTable;
-import com.datasqrl.plan.table.VirtualRelationalTable;
+import com.datasqrl.plan.table.PhysicalTable;
 import com.datasqrl.plan.global.PhysicalDAGPlan.EngineSink;
 import com.datasqrl.plan.global.SqrlDAG.ExportNode;
 import com.datasqrl.plan.local.generate.Debugger;
@@ -74,7 +74,7 @@ public PhysicalDAGPlan assemble(SqrlDAG dag, Set<URL> jars, Map<String, UserDefi
     dag.allNodesByClass(SqrlDAG.TableNode.class).forEach( tableNode -> {
       ExecutionStage stage = tableNode.getChosenStage();
       Preconditions.checkNotNull(stage);
-      ScriptTable table = tableNode.getTable();
+      PhysicalTable table = tableNode.getTable();
       table.assignStage(stage); //this stage on the config below
       SQRLConverter.Config config = table.getBaseConfig().build();
       table.setPlannedRelNode(sqrlConverter.convert(table, config, errors));
@@ -126,16 +126,16 @@ public PhysicalDAGPlan assemble(SqrlDAG dag, Set<URL> jars, Map<String, UserDefi
       Preconditions.checkArgument(tableScanVisitor.scanFunctions.isEmpty(),
           "Should not encounter table functions in materialized queries");
       Set<AbstractRelationalTable> materializedTables = tableScanVisitor.scanTables;
-      List<VirtualRelationalTable> normalizedTables = StreamUtil.filterByClass(materializedTables,
-          VirtualRelationalTable.class).sorted().collect(Collectors.toList());
+      List<LogicalNestedTable> normalizedTables = StreamUtil.filterByClass(materializedTables,
+              LogicalNestedTable.class).sorted().collect(Collectors.toList());
       List<PhysicalRelationalTable> denormalizedTables = StreamUtil.filterByClass(materializedTables,
           PhysicalRelationalTable.class).sorted().collect(Collectors.toList());
 
       //Fill all table sinks
       //First, all the tables that need to be written to the database in normalized form
-      for (VirtualRelationalTable normTable : normalizedTables) {
+      for (LogicalNestedTable normTable : normalizedTables) {
         RelNode scanTable = sqrlConverter.getRelBuilder().scan(normTable.getNameId()).build();
-        SQRLConverter.Config.ConfigBuilder configBuilder = normTable.getRoot().getBase().getBaseConfig();
+        SQRLConverter.Config.ConfigBuilder configBuilder = normTable.getRoot().getBaseConfig();
         configBuilder.fieldNames(normTable.getRowType().getFieldNames());
         Pair<RelNode, Integer> relPlusTimestamp = produceWriteTree(scanTable,
             configBuilder.build(), errors);
@@ -190,7 +190,7 @@ public PhysicalDAGPlan assemble(SqrlDAG dag, Set<URL> jars, Map<String, UserDefi
     Lists.newArrayList(Iterables.filter(dag, SqrlDAG.TableNode.class)).stream()
         .filter(node -> node.getChosenStage().equals(streamStage))
         .map(SqrlDAG.TableNode::getTable).filter(tbl -> debugger.isDebugTable(tbl.getTableName()))
-        .sorted(Comparator.comparing(ScriptTable::getTableName))
+        .sorted(Comparator.comparing(PhysicalTable::getTableName))
         .forEach(table -> {
           Name debugSinkName = table.getTableName().suffix("debug" + debugCounter.incrementAndGet());
           TableSink sink = debugger.getDebugSink(debugSinkName, errors);
@@ -247,7 +247,7 @@ private Pair<RelNode,Integer> produceWriteTree(RelNode relNode, SQRLConverter.Co
   private static class VisitTableScans extends RelShuttleImpl {
 
     final Set<AbstractRelationalTable> scanTables = new HashSet<>();
-    final Set<TableFunctionBase> scanFunctions = new HashSet<>();
+    final Set<QueryTableFunction> scanFunctions = new HashSet<>();
 
     public void findScans(RelNode relNode) {
       relNode.accept(this);
@@ -257,7 +257,7 @@ public void findScans(RelNode relNode) {
     public RelNode visit(TableScan scan) {
       PhysicalRelationalTable table = scan.getTable().unwrap(PhysicalRelationalTable.class);
       if (table == null) { //It's a normalized query
-        VirtualRelationalTable vtable = scan.getTable().unwrap(VirtualRelationalTable.class);
+        LogicalNestedTable vtable = scan.getTable().unwrap(LogicalNestedTable.class);
         Preconditions.checkNotNull(vtable);
         scanTables.add(vtable);
       } else {
@@ -268,8 +268,8 @@ public RelNode visit(TableScan scan) {
 
     @Override
     public RelNode visit(TableFunctionScan scan) {
-      SqrlRexUtil.getCustomTableFunction(scan).filter(TableFunctionBase.class::isInstance)
-          .map(TableFunctionBase.class::cast).ifPresent(scanFunctions::add);
+      SqrlRexUtil.getCustomTableFunction(scan).filter(QueryTableFunction.class::isInstance)
+          .map(QueryTableFunction.class::cast).ifPresent(scanFunctions::add);
       return super.visit(scan);
     }
   }

diff --git a/sqrl-planner/sqrl-planner-local/src/main/java/com/datasqrl/plan/global/DAGBuilder.java b/sqrl-planner/sqrl-planner-local/src/main/java/com/datasqrl/plan/global/DAGBuilder.java
@@ -16,7 +16,8 @@
 import com.datasqrl.plan.global.SqrlDAG.TableNode;
 import com.datasqrl.plan.global.StageAnalysis.Cost;
 import com.datasqrl.plan.local.generate.ResolvedExport;
-import com.datasqrl.plan.table.ScriptTable;
+import com.datasqrl.plan.table.PhysicalRelationalTable;
+import com.datasqrl.plan.table.PhysicalTable;
 import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Multimap;
 import java.util.Collection;
@@ -46,7 +47,7 @@ public class DAGBuilder {
 
   public SqrlDAG build(Collection<AnalyzedAPIQuery> queries,
       Collection<ResolvedExport> exports) {
-    Map<ScriptTable, TableNode> table2Node = new HashMap<>();
+    Map<PhysicalRelationalTable, TableNode> table2Node = new HashMap<>();
     Multimap<SqrlNode, SqrlNode> dagInputs = HashMultimap.create();
     //1. Add all queries as sinks
     List<ExecutionStage> readStages = pipeline.getReadStages();
@@ -73,8 +74,8 @@ public SqrlDAG build(Collection<AnalyzedAPIQuery> queries,
   private void add2DAG(RelNode relnode, Config baseConfig,
       List<ExecutionStage> stages, Multimap<SqrlNode, SqrlNode> dagInputs,
       Function<Map<ExecutionStage, StageAnalysis>,SqrlNode> nodeConstructor,
-      Map<ScriptTable, TableNode> table2Node) {
-    Set<ScriptTable> inputTables = new LinkedHashSet<>();
+      Map<PhysicalRelationalTable, TableNode> table2Node) {
+    Set<PhysicalRelationalTable> inputTables = new LinkedHashSet<>();
     Config.ConfigBuilder configBuilder = baseConfig.toBuilder()
         .sourceTableConsumer(inputTables::add);
 
@@ -87,11 +88,11 @@ private void add2DAG(RelNode relnode, Config baseConfig,
         dagInputs.put(node,input));
   }
 
-  private SqrlDAG.TableNode getInputTable(ScriptTable table,
-      Multimap<SqrlNode, SqrlNode> dagInputs,
-      Map<ScriptTable, TableNode> table2Node) {
+  private SqrlDAG.TableNode getInputTable(PhysicalRelationalTable table,
+                                          Multimap<SqrlNode, SqrlNode> dagInputs,
+                                          Map<PhysicalRelationalTable, TableNode> table2Node) {
     if (table2Node.containsKey(table)) return table2Node.get(table);
-    Set<ScriptTable> inputTables = new LinkedHashSet<>();
+    Set<PhysicalRelationalTable> inputTables = new LinkedHashSet<>();
     SQRLConverter.Config.ConfigBuilder configBuilder = table.getBaseConfig();
     configBuilder.sourceTableConsumer(inputTables::add);
     List<ExecutionStage> stages = table.getSupportedStages(pipeline, errors);
@@ -105,7 +106,7 @@ private SqrlDAG.TableNode getInputTable(ScriptTable table,
     return node;
   }
 
-  public Map<ExecutionStage, StageAnalysis> planStages(ScriptTable table) {
+  public Map<ExecutionStage, StageAnalysis> planStages(PhysicalTable table) {
     SQRLConverter.Config.ConfigBuilder configBuilder = table.getBaseConfig();
     List<ExecutionStage> stages = table.getSupportedStages(pipeline, errors);
     return tryStages(stages, stage ->

diff --git a/sqrl-planner/sqrl-planner-local/src/main/java/com/datasqrl/plan/global/DAGPreparation.java b/sqrl-planner/sqrl-planner-local/src/main/java/com/datasqrl/plan/global/DAGPreparation.java
@@ -5,11 +5,11 @@
 import com.datasqrl.calcite.SqrlFramework;
 import com.datasqrl.error.ErrorCollector;
 import com.datasqrl.graphql.APIConnectorManager;
-import com.datasqrl.plan.local.generate.ComputeTableFunction;
+import com.datasqrl.plan.local.generate.QueryTableFunction;
 import com.datasqrl.plan.local.generate.ResolvedExport;
+import com.datasqrl.plan.table.LogicalNestedTable;
 import com.datasqrl.plan.table.ProxyImportRelationalTable;
 import com.datasqrl.plan.table.PhysicalRelationalTable;
-import com.datasqrl.plan.table.VirtualRelationalTable;
 import com.datasqrl.util.StreamUtil;
 import com.google.common.base.Preconditions;
 import java.util.Collection;
@@ -46,7 +46,7 @@ public Collection<AnalyzedAPIQuery> prepareInputs(SqrlSchema sqrlSchema, APIConn
 
     //Append timestamp column to nested, normalized tables, so we can propagate it
     //to engines that don't support denormalization
-    sqrlSchema.getTableStream(VirtualRelationalTable.Child.class)
+    sqrlSchema.getTableStream(LogicalNestedTable.class)
         .forEach(nestedTable -> {
           nestedTable.appendTimestampColumn(relBuilder.getTypeFactory());
         });
@@ -59,8 +59,8 @@ public Collection<AnalyzedAPIQuery> prepareInputs(SqrlSchema sqrlSchema, APIConn
 
   private Stream<PhysicalRelationalTable> getAllPhysicalTables(SqrlSchema sqrlSchema) {
     return Stream.concat(sqrlSchema.getTables(PhysicalRelationalTable.class).stream(),
-            sqrlSchema.getFunctionStream(ComputeTableFunction.class).map(
-                    ComputeTableFunction::getQueryTable));
+            sqrlSchema.getFunctionStream(QueryTableFunction.class).map(
+                    QueryTableFunction::getQueryTable));
   }
 
   private void finalizeTimestampOnTable(PhysicalRelationalTable table) {

diff --git a/sqrl-planner/sqrl-planner-local/src/main/java/com/datasqrl/plan/global/DatabaseQuery.java b/sqrl-planner/sqrl-planner-local/src/main/java/com/datasqrl/plan/global/DatabaseQuery.java
@@ -2,13 +2,12 @@
 
 import com.datasqrl.engine.pipeline.ExecutionStage;
 import com.datasqrl.error.ErrorCollector;
-import com.datasqrl.plan.local.generate.AccessTableFunction;
-import com.datasqrl.plan.local.generate.ComputeTableFunction;
-import com.datasqrl.plan.local.generate.TableFunctionBase;
+import com.datasqrl.plan.local.generate.QueryTableFunction;
 import com.datasqrl.plan.queries.IdentifiedQuery;
 import com.datasqrl.plan.rules.SQRLConverter;
 import com.datasqrl.plan.table.AbstractRelationalTable;
-import com.datasqrl.plan.table.VirtualRelationalTable;
+import com.datasqrl.plan.table.PhysicalRelationalTable;
+import com.datasqrl.plan.table.QueryRelationalTable;
 import com.google.common.base.Preconditions;
 import lombok.Value;
 import org.apache.calcite.rel.RelNode;
@@ -22,25 +21,19 @@ default String getName() {
   }
 
   static DatabaseQuery.Instance of(AbstractRelationalTable table) {
-    Preconditions.checkArgument(table instanceof VirtualRelationalTable.Root, "Expected root virtual table");
-    VirtualRelationalTable.Root vTable = (VirtualRelationalTable.Root)table;
+    Preconditions.checkArgument(table instanceof PhysicalRelationalTable, "Expected physical table");
+    PhysicalRelationalTable vTable = (PhysicalRelationalTable)table;
     Preconditions.checkArgument(vTable.isRoot());
-    ExecutionStage stage = vTable.getRoot().getBase().getAssignedStage().get();
+    ExecutionStage stage = vTable.getAssignedStage().get();
     //TODO: We don't yet support server queries directly against materialized tables. Need a database stage in between.
     Preconditions.checkArgument(stage.isRead(), "We do not yet support queries directly against stream");
-    return new Instance(vTable.getNameId(), vTable.getRoot().getBase().getPlannedRelNode(), stage);
+    return new Instance(vTable.getNameId(), vTable.getPlannedRelNode(), stage);
   }
 
-  static DatabaseQuery.Instance of(TableFunctionBase function) {
-    ExecutionStage assignedStage;
-    if (function instanceof AccessTableFunction) {
-      AccessTableFunction accessFct = (AccessTableFunction) function;
-      assignedStage = accessFct.getAssignedStage().get();
-    } else {
-      ComputeTableFunction computeFct = (ComputeTableFunction) function;
-      assignedStage = computeFct.getQueryTable().getAssignedStage().get();
-    }
-    return new Instance(function.getNameId(), function.getPlannedRelNode(), assignedStage);
+  static DatabaseQuery.Instance of(QueryTableFunction function) {
+    QueryRelationalTable queryTable = function.getQueryTable();
+    ExecutionStage assignedStage = queryTable.getAssignedStage().get();
+    return new Instance(queryTable.getNameId(), queryTable.getPlannedRelNode(), assignedStage);
   }