Refactor TableSource to store columnNames for better SQL projection

Author: tanyarai5446

wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/TableSource.java

@@ -29,6 +29,12 @@
 public class TableSource extends UnarySource<Record> {
 
     private final String tableName;
+    private final String[] columnNames; // Line 32
+
+    // Add this method somewhere around line 35
+    public String[] getColumnNames() {
+        return this.columnNames;
+    }
 
     public String getTableName() {
         return tableName;
@@ -42,12 +48,15 @@ public String getTableName() {
      *                    into Wayang, so as to allow specific optimizations
      */
     public TableSource(String tableName, String... columnNames) {
-        this(tableName, createOutputDataSetType(columnNames));
+        super(createOutputDataSetType(columnNames));
+        this.tableName = tableName;
+        this.columnNames = columnNames; // Assigned here!
     }
 
     public TableSource(String tableName, DataSetType<Record> type) {
         super(type);
         this.tableName = tableName;
+        this.columnNames=new String[0];
     }
 
     /**
@@ -71,6 +80,7 @@ private static DataSetType<Record> createOutputDataSetType(String[] columnNames)
     public TableSource(TableSource that) {
         super(that);
         this.tableName = that.getTableName();
+        this.columnNames = that.getColumnNames(); // ADD THIS LINE HERE!
     }
 
 }

wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcTableSource.java

@@ -55,9 +55,16 @@ public JdbcTableSource(JdbcTableSource that) {
 
     @Override
     public String createSqlClause(Connection connection, FunctionCompiler compiler) {
-        return this.getTableName();
-    }
+        // Call the method we just created in the parent
+        String[] cols = this.getColumnNames();
+
+        if (cols == null || cols.length == 0) {
+            return String.format("SELECT * FROM %s", this.getTableName());
+        }
 
+        String columns = String.join(", ", cols);
+        return String.format("SELECT %s FROM %s", columns, this.getTableName());
+    }
 
     @Override
     public String getLoadProfileEstimatorConfigurationKey() {