Add DuckDB platform support with Parquet source and cost model

wayang-applications/README.md

@@ -76,4 +76,7 @@ The file _env.demo1.sh_ contains additional properties, need in a particlar demo
 In this file we will never see cluster or user specific details, only properties which are specific to the 
 particular application are listed here.
 
+## DuckDB Example
 
+The [DuckDB example](duckdb.md) demonstrates filter and projection plans against
+a configurable DuckDB database.

wayang-applications/duckdb.md

@@ -0,0 +1,70 @@
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements. See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+-->
+
+# DuckDB Example
+
+`org.apache.wayang.applications.DuckDBDemo` runs filter and projection plans
+through Wayang against an embedded DuckDB database. Java 17 is required. No
+Docker installation or database server is needed.
+
+Create a properties file, for example `/tmp/duckdb-example.properties`:
+
+```properties
+wayang.duckdb.jdbc.url = jdbc:duckdb:/tmp/wayang-example.duckdb
+wayang.duckdb.demo.orders = wayang_demo.orders
+wayang.duckdb.demo.filter-result = wayang_demo.filter_result
+wayang.duckdb.demo.projection-result = wayang_demo.projection_result
+```
+
+Use an absolute database path with an existing parent directory. The example
+uses multiple JDBC connections, so it requires a persistent file. On Windows,
+use a path such as `C:/Temp/wayang-example.duckdb`.
+
+From the repository root, build and install the application and its dependencies:
+
+```bash
+./mvnw -Pskip-prerequisite-check -pl wayang-applications -am \
+  -DskipTests -Dpython.worker.tests.skip=true install
+```
+
+To create sample data and run the plans:
+
+```bash
+./mvnw -Pskip-prerequisite-check -pl wayang-applications exec:java \
+  -Dexec.mainClass=org.apache.wayang.applications.DuckDBDemo \
+  "-Dexec.args=file:///tmp/duckdb-example.properties --init"
+```
+
+The `--init` option creates six sample orders. It fails if the input table
+already exists, so it cannot replace existing input data. If a different input
+schema is configured, create that schema before initialization.
+
+To use an existing table, omit `--init`:
+
+```bash
+./mvnw -Pskip-prerequisite-check -pl wayang-applications exec:java \
+  -Dexec.mainClass=org.apache.wayang.applications.DuckDBDemo \
+  "-Dexec.args=file:///tmp/duckdb-example.properties"
+```
+
+The input table must contain `order_id BIGINT`, `customer_id BIGINT`,
+`region VARCHAR`, and `amount DOUBLE`. Configure distinct input and output table
+names, and create any custom output schemas before running the example. Each run
+replaces the two configured output tables.
+
+On Windows, replace `./mvnw` with `.\mvnw.cmd` and use a configuration URL such
+as `file:///C:/Temp/duckdb-example.properties`.

wayang-applications/pom.xml

@@ -56,6 +56,16 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-duckdb</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.antlr</groupId>
+            <artifactId>antlr4-runtime</artifactId>
+            <version>4.13.1</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.wayang</groupId>
             <artifactId>wayang-core</artifactId>
@@ -104,6 +114,11 @@
             <version>3.9.2</version> <!-- Use the latest version available -->
         </dependency>
 
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <version>2.18.8</version>
+        </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>

wayang-applications/src/main/java/org/apache/wayang/applications/DuckDBDemo.java

@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.applications;
+
+import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.function.ProjectionDescriptor;
+import org.apache.wayang.basic.operators.FilterOperator;
+import org.apache.wayang.basic.operators.MapOperator;
+import org.apache.wayang.basic.operators.TableSink;
+import org.apache.wayang.basic.types.RecordType;
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.api.WayangContext;
+import org.apache.wayang.core.function.PredicateDescriptor;
+import org.apache.wayang.core.plan.wayangplan.WayangPlan;
+import org.apache.wayang.core.types.DataSetType;
+import org.apache.wayang.duckdb.DuckDB;
+import org.apache.wayang.duckdb.operators.DuckDBTableSource;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Properties;
+
+/**
+ * Configurable DuckDB filter and projection example.
+ * See {@code wayang-applications/duckdb.md} for usage and fixture initialization.
+ */
+public class DuckDBDemo {
+
+    private final Configuration configuration;
+    private final String jdbcUrl;
+    private final String orders;
+    private final String filterResult;
+    private final String projectionResult;
+
+    private DuckDBDemo(Configuration configuration) {
+        this.configuration = configuration;
+        this.jdbcUrl = configuration.getStringProperty("wayang.duckdb.jdbc.url");
+        if ("jdbc:duckdb:".equals(jdbcUrl)) {
+            throw new IllegalArgumentException("This example requires a database file shared by its JDBC connections.");
+        }
+        this.orders = configuration.getStringProperty("wayang.duckdb.demo.orders", "wayang_demo.orders");
+        this.filterResult = configuration.getStringProperty(
+                "wayang.duckdb.demo.filter-result", "wayang_demo.filter_result"
+        );
+        this.projectionResult = configuration.getStringProperty(
+                "wayang.duckdb.demo.projection-result", "wayang_demo.projection_result"
+        );
+        if (orders.equalsIgnoreCase(filterResult) || orders.equalsIgnoreCase(projectionResult)
+                || filterResult.equalsIgnoreCase(projectionResult)) {
+            throw new IllegalArgumentException("Input and output tables must have distinct names.");
+        }
+    }
+
+    public static void main(String[] args) throws Exception {
+        if (args.length < 1 || args.length > 2 || (args.length == 2 && !"--init".equals(args[1]))) {
+            throw new IllegalArgumentException("Usage: DuckDBDemo <configuration URL> [--init]");
+        }
+        DuckDBDemo demo = new DuckDBDemo(new Configuration(args[0]));
+        if (args.length == 2) {
+            demo.createFixture();
+        }
+        demo.runFilterPushdown();
+        demo.runProjectionPushdown();
+    }
+
+    private void runFilterPushdown() throws Exception {
+        System.out.println();
+        System.out.println("DuckDB demo: Filter pushdown");
+        System.out.println("SQL shape: SELECT * FROM " + orders + " WHERE region = 'AMER'");
+
+        DuckDBTableSource source = new DuckDBTableSource(
+                orders, "order_id", "customer_id", "region", "amount");
+        FilterOperator<Record> filter = new FilterOperator<>(
+                new PredicateDescriptor<>(
+                        (Record record) -> "AMER".equals(record.getField(2)), Record.class)
+                        .withSqlImplementation("region = 'AMER'"));
+        TableSink<Record> sink = new TableSink<>(
+                new Properties(), "overwrite", filterResult,
+                "order_id", "customer_id", "region", "amount");
+
+        source.connectTo(0, filter, 0);
+        filter.connectTo(0, sink, 0);
+        wayangContext().execute("DuckDB filter demo", new WayangPlan(sink));
+
+        printQuery("SELECT order_id, region, amount FROM " + filterResult + " ORDER BY order_id");
+    }
+
+    private void runProjectionPushdown() throws Exception {
+        System.out.println();
+        System.out.println("DuckDB demo: Projection + filter pushdown");
+        System.out.println("SQL shape: SELECT region, amount FROM " + orders + " WHERE region = 'AMER'");
+
+        DuckDBTableSource source = new DuckDBTableSource(
+                orders, "order_id", "customer_id", "region", "amount");
+        FilterOperator<Record> filter = new FilterOperator<>(
+                new PredicateDescriptor<>(
+                        (Record record) -> "AMER".equals(record.getField(2)), Record.class)
+                        .withSqlImplementation("region = 'AMER'"));
+        MapOperator<Record, Record> projection = new MapOperator<>(
+                ProjectionDescriptor.createForRecords(
+                        new RecordType("order_id", "customer_id", "region", "amount"),
+                        "region", "amount"),
+                DataSetType.createDefault(Record.class),
+                DataSetType.createDefault(Record.class));
+        TableSink<Record> sink = new TableSink<>(
+                new Properties(), "overwrite", projectionResult,
+                "region", "amount");
+
+        source.connectTo(0, filter, 0);
+        filter.connectTo(0, projection, 0);
+        projection.connectTo(0, sink, 0);
+        wayangContext().execute("DuckDB projection demo", new WayangPlan(sink));
+
+        printQuery("SELECT region, amount FROM " + projectionResult + " ORDER BY amount DESC");
+    }
+
+    private WayangContext wayangContext() {
+        return new WayangContext(configuration)
+                .withPlugin(DuckDB.plugin());
+    }
+
+    private void createFixture() throws Exception {
+        try (Connection connection = DriverManager.getConnection(jdbcUrl);
+             Statement statement = connection.createStatement()) {
+            statement.execute("CREATE SCHEMA IF NOT EXISTS wayang_demo");
+            statement.execute("CREATE TABLE " + orders + " ("
+                    + "order_id BIGINT, customer_id BIGINT, region VARCHAR, amount DOUBLE)");
+            statement.execute("INSERT INTO " + orders + " VALUES "
+                    + "(1, 100, 'AMER', 2200.0),"
+                    + "(2, 101, 'EMEA',  800.5),"
+                    + "(3, 100, 'AMER',  680.5),"
+                    + "(4, 102, 'APAC', 1500.0),"
+                    + "(5, 101, 'EMEA', 1100.0),"
+                    + "(6, 100, 'AMER',  950.25)");
+        }
+    }
+
+    private void printQuery(String sql) throws Exception {
+        try (Connection connection = DriverManager.getConnection(jdbcUrl);
+             Statement statement = connection.createStatement();
+             ResultSet resultSet = statement.executeQuery(sql)) {
+            int columns = resultSet.getMetaData().getColumnCount();
+            while (resultSet.next()) {
+                StringBuilder row = new StringBuilder("  ");
+                for (int i = 1; i <= columns; i++) {
+                    if (i > 1) {
+                        row.append(" | ");
+                    }
+                    row.append(resultSet.getObject(i));
+                }
+                System.out.println(row);
+            }
+        }
+    }
+}

wayang-applications/src/test/java/org/apache/wayang/applications/DuckDBDemoTest.java

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.applications;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.DriverManager;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class DuckDBDemoTest {
+    @TempDir
+    Path directory;
+
+    @Test
+    void initializesOnlyOnRequestAndUsesExistingData() throws Exception {
+        String jdbcUrl = "jdbc:duckdb:" + directory.resolve("example.duckdb");
+        Path config = directory.resolve("example.properties");
+        Files.writeString(config, "wayang.duckdb.jdbc.url = " + jdbcUrl.replace("\\", "/") + "\n");
+        String configUrl = config.toUri().toString();
+        DuckDBDemo.main(new String[]{configUrl, "--init"});
+        try (var connection = DriverManager.getConnection(jdbcUrl);
+             var statement = connection.createStatement()) {
+            statement.execute("INSERT INTO wayang_demo.orders VALUES (7, 103, 'AMER', 42.0)");
+        }
+        DuckDBDemo.main(new String[]{configUrl});
+        try (var connection = DriverManager.getConnection(jdbcUrl);
+             var statement = connection.createStatement()) {
+            try (var result = statement.executeQuery("SELECT count(*) FROM wayang_demo.orders")) {
+                assertTrue(result.next());
+                assertEquals(7, result.getInt(1));
+            }
+            try (var result = statement.executeQuery(
+                    "SELECT count(*), sum(amount) FROM wayang_demo.projection_result"
+            )) {
+                assertTrue(result.next());
+                assertEquals(4, result.getInt(1));
+                assertEquals(3872.75, result.getDouble(2), 0.001);
+            }
+        }
+        assertThrows(Exception.class, () -> DuckDBDemo.main(new String[]{configUrl, "--init"}));
+        try (var connection = DriverManager.getConnection(jdbcUrl);
+             var statement = connection.createStatement();
+             var result = statement.executeQuery("SELECT count(*) FROM wayang_demo.orders")) {
+            assertTrue(result.next());
+            assertEquals(7, result.getInt(1));
+        }
+    }
+}

wayang-platforms/pom.xml

@@ -46,6 +46,7 @@
         <module>wayang-bigquery</module>
         <module>wayang-presto</module>
         <module>wayang-trino</module>
+        <module>wayang-duckdb</module>
         <module>wayang-tensorflow</module>
     </modules>