修复aggtable

dlink-core/src/main/java/com/dlink/explainer/Explainer.java

@@ -75,21 +75,22 @@ public class Explainer {
         List<TableCAResult> results = new ArrayList<>();
         for (int i = 0; i < strPlans.size(); i++) {
             List<Trans> trans = translateTrans(translateObjectNode(strPlans.get(i)));
-            TableCAGenerator generator = new TableCAGenerator(trans);
+            TableCAGenerator generator = TableCAGenerator.build(trans);
             if (onlyTable) {
                 generator.translateOnlyTable();
             } else {
                 generator.translate();
             }
-            results.add(new TableCAResult(generator));
+            results.add(generator.getResult());
         }
         if (results.size() > 0) {
             CatalogManager catalogManager = executor.getCatalogManager();
             for (int i = 0; i < results.size(); i++) {
                 TableCA sinkTableCA = (TableCA) results.get(i).getSinkTableCA();
-                if (sinkTableCA != null) {
-                    ObjectIdentifier objectIdentifier = ObjectIdentifier.of(sinkTableCA.getCatalog(), sinkTableCA.getDatabase(), sinkTableCA.getTable());
-                    Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable(objectIdentifier);
+                if (Asserts.isNotNull(sinkTableCA)) {
+                    Optional<CatalogManager.TableLookupResult> tableOpt = catalogManager.getTable(
+                            ObjectIdentifier.of(sinkTableCA.getCatalog(), sinkTableCA.getDatabase(), sinkTableCA.getTable())
+                    );
                     if (tableOpt.isPresent()) {
                         String[] fieldNames = tableOpt.get().getResolvedSchema().getFieldNames();
                         sinkTableCA.setFields(Arrays.asList(fieldNames));

dlink-core/src/main/java/com/dlink/explainer/ca/TableCA.java

@@ -2,6 +2,7 @@ package com.dlink.explainer.ca;
 
 import com.dlink.explainer.trans.SinkTrans;
 import com.dlink.explainer.trans.SourceTrans;
+import com.dlink.explainer.trans.Trans;
 import lombok.Getter;
 import lombok.Setter;
 
@@ -31,6 +32,11 @@ public class TableCA implements ICA{
     private Set<Integer> columnCAIds = new HashSet<>();
     private Integer parallelism;
 
+    private static final TableCA EMPTY = new TableCA();
+
+    public TableCA() {
+    }
+
     public TableCA(SourceTrans trans) {
         this.id = trans.getId();
         this.parentId = trans.getParentId();
@@ -57,6 +63,16 @@ public class TableCA implements ICA{
         this.type = trans.getPact();
     }
 
+    public static TableCA build(Trans trans){
+        if(trans instanceof SourceTrans){
+            return new TableCA((SourceTrans)trans);
+        }else if(trans instanceof SinkTrans){
+            return new TableCA((SinkTrans)trans);
+        }else{
+            return TableCA.EMPTY;
+        }
+    }
+
     @Override
     public String toString() {
         return "TableCA{" +

dlink-core/src/main/java/com/dlink/explainer/ca/TableCAGenerator.java

@@ -44,11 +44,19 @@ public class TableCAGenerator implements CAGenerator {
         }
     }
 
+    public static TableCAGenerator build(List<Trans> transList) {
+        return new TableCAGenerator(transList);
+    }
+
+    public TableCAResult getResult(){
+        return new TableCAResult(this);
+    }
+
     @Override
     public void translate() {
-        for (int i = 0; i < transList.size(); i++) {
-            if(transList.get(i) instanceof SourceTrans) {
-                TableCA tableCA = new TableCA((SourceTrans) transList.get(i));
+        for(Trans trans : transList){
+            if(trans instanceof SourceTrans) {
+                TableCA tableCA = TableCA.build(trans);
                 List<String> sourceFields = new ArrayList<>();
                 CollectionUtils.addAll(sourceFields, new Object[tableCA.getFields().size()]);
                 Collections.copy(sourceFields, tableCA.getFields());
@@ -56,8 +64,8 @@ public class TableCAGenerator implements CAGenerator {
                     buildTableCAFields(tableCA,tableCA.getParentId(),sourceFields.get(j));
                 }
                 this.sourceTableCAS.add(tableCA);
-            }else if(transList.get(i) instanceof SinkTrans) {
-                TableCA tableCA = new TableCA((SinkTrans) transList.get(i));
+            }else if(trans instanceof SinkTrans) {
+                TableCA tableCA = TableCA.build(trans);
                 this.sinkTableCA = tableCA;
                 this.sinkTableName = tableCA.getName();
             }
@@ -65,11 +73,11 @@ public class TableCAGenerator implements CAGenerator {
     }
 
     public void translateOnlyTable() {
-        for (int i = 0; i < transList.size(); i++) {
-            if(transList.get(i) instanceof SourceTrans) {
-                this.sourceTableCAS.add(new TableCA((SourceTrans) transList.get(i)));
-            }else if(transList.get(i) instanceof SinkTrans) {
-                TableCA tableCA = new TableCA((SinkTrans) transList.get(i));
+        for(Trans trans : transList){
+            if(trans instanceof SourceTrans) {
+                this.sourceTableCAS.add(new TableCA((SourceTrans) trans));
+            }else if(trans instanceof SinkTrans) {
+                TableCA tableCA = new TableCA((SinkTrans) trans);
                 this.sinkTableCA = tableCA;
                 this.sinkTableName = tableCA.getName();
             }

dlink-core/src/main/java/com/dlink/explainer/trans/TransGenerator.java

@@ -59,7 +59,7 @@ public class TransGenerator {
         for (Map.Entry<Integer, Trans> entry : nodemap.entrySet()) {
             Trans trans = entry.getValue();
             List<Predecessor> predecessors = trans.getPredecessors();
-            if (predecessors == null || predecessors.size() == 0) {
+            if (Asserts.isNull(predecessors)) {
                 continue;
             }
             for (int i = 0; i < predecessors.size(); i++) {

dlink-core/src/main/java/com/dlink/parser/SingleSqlParserFactory.java

@@ -16,7 +16,7 @@ public class SingleSqlParserFactory {
     public static Map<String,List<String>> generateParser(String sql) {
         BaseSingleSqlParser tmp = null;
 //        sql = sql.replace("\n"," ").replaceAll("\\s{1,}", " ") +" ENDOFSQL";
-        sql = sql.replace("\n"," ") +" ENDOFSQL";
+        sql = sql.replace("\r\n"," ").replace("\n"," ") +" ENDOFSQL";
         if (contains(sql, "(insert\\s+into)(.+)(select)(.+)(from)(.+)")) {
             tmp = new InsertSelectSqlParser(sql);
         } else if (contains(sql, "(create\\s+aggtable)(.+)(as\\s+select)(.+)")) {

dlink-core/src/test/java/com/dlink/core/SqlParserTest.java

@@ -30,8 +30,14 @@ public class SqlParserTest {
                 "WHERE cls = 1\n" +
                 "GROUP BY sid\n" +
                 "AGG BY toMap(cls,score) as (data)";
+        String sql2 = "\r\n" +
+                "CREATE AGGTABLE aggscore AS \r\n" +
+                "SELECT cls,score,rank\r\n" +
+                "FROM score\r\n" +
+                "GROUP BY cls\r\n" +
+                "AGG BY TOP2(score) as (score,rank)";
         //sql=sql.replace("\n"," ");
-        Map<String,List<String>> lists = SingleSqlParserFactory.generateParser(sql);
+        Map<String,List<String>> lists = SingleSqlParserFactory.generateParser(sql2);
         System.out.println(lists.toString());
         System.out.println(StringUtils.join(lists.get("SELECT"),","));
     }