[WAYANG-55] Refactor TableSource to support column projection.

python/src/pywy/tests/resources/sample_data.md

@@ -246,7 +246,7 @@ The list of [contributors](https://github.com/apache/incubator-wayang/graphs/con
 ## License
 All files in this repository are licensed under the Apache Software License 2.0
 
-Copyright 2020 - 2025 The Apache Software Foundation.
+Copyright 2020 - 2026 The Apache Software Foundation.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

wayang-applications/data/case-study/DATA_REPO_001/README.md

@@ -207,7 +207,7 @@ The list of [contributors](https://github.com/apache/incubator-wayang/graphs/con
 ## License
 All files in this repository are licensed under the Apache Software License 2.0
 
-Copyright 2020 - 2024 The Apache Software Foundation.
+Copyright 2020 - 2026 The Apache Software Foundation.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

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-docs/src/main/resources/index.md

@@ -387,7 +387,7 @@ object kmeans {
 
 All files in this repository are licensed under the Apache Software License 2.0
 
-Copyright 2020 - 2024 The Apache Software Foundation.
+Copyright 2020 - 2026 The Apache Software Foundation.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

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() {