that) {
+ super(that);
+ }
+
+ @Override
+ protected DuckDBSortOperator createCopy() {
+ return new DuckDBSortOperator(this);
+ }
+}
diff --git a/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/operators/DuckDBTableSinkOperator.java b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/operators/DuckDBTableSinkOperator.java
new file mode 100644
index 000000000..35977c920
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/operators/DuckDBTableSinkOperator.java
@@ -0,0 +1,48 @@
+/*
+ * 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.duckdb.operators;
+
+import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.operators.TableSink;
+import org.apache.wayang.jdbc.operators.JdbcTableSinkOperator;
+
+/**
+ * DuckDB implementation of the {@link JdbcTableSinkOperator}. The sink stays
+ * entirely within DuckDB: the composed query is wrapped in a
+ * {@code CREATE TABLE ... AS} (mode {@code overwrite}) or {@code INSERT INTO ...}
+ * statement.
+ *
+ * Table names can be unqualified or schema-qualified, e.g.
+ * {@code orders} or {@code main.orders}.
+ */
+public class DuckDBTableSinkOperator extends JdbcTableSinkOperator implements DuckDBExecutionOperator {
+
+ public DuckDBTableSinkOperator(String tableName, String[] columnNames) {
+ super(tableName, columnNames);
+ }
+
+ /**
+ * Copies an instance (exclusive of broadcasts).
+ *
+ * @param that that should be copied
+ */
+ public DuckDBTableSinkOperator(TableSink that) {
+ super(that);
+ }
+}
diff --git a/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/operators/DuckDBTableSource.java b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/operators/DuckDBTableSource.java
new file mode 100644
index 000000000..47eab390d
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/operators/DuckDBTableSource.java
@@ -0,0 +1,55 @@
+/*
+ * 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.duckdb.operators;
+
+import org.apache.wayang.basic.operators.TableSource;
+import org.apache.wayang.core.platform.ChannelDescriptor;
+import org.apache.wayang.jdbc.operators.JdbcTableSource;
+
+import java.util.List;
+
+/**
+ * DuckDB implementation of the {@link TableSource}.
+ *
+ * Table names can be unqualified or schema-qualified, e.g.
+ * {@code orders} or {@code main.orders}.
+ */
+public class DuckDBTableSource extends JdbcTableSource implements DuckDBExecutionOperator {
+
+ /**
+ * @see TableSource#TableSource(String, String...)
+ */
+ public DuckDBTableSource(String tableName, String... columnNames) {
+ super(tableName, columnNames);
+ }
+
+ /**
+ * Copies an instance (exclusive of broadcasts).
+ *
+ * @param that that should be copied
+ */
+ public DuckDBTableSource(JdbcTableSource that) {
+ super(that);
+ }
+
+ @Override
+ public List getSupportedInputChannels(int index) {
+ throw new UnsupportedOperationException("This operator has no input channels.");
+ }
+}
diff --git a/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/platform/DuckDBPlatform.java b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/platform/DuckDBPlatform.java
new file mode 100644
index 000000000..8ad24135d
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/platform/DuckDBPlatform.java
@@ -0,0 +1,54 @@
+/*
+ * 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.duckdb.platform;
+
+import org.apache.wayang.core.platform.Platform;
+import org.apache.wayang.jdbc.platform.JdbcPlatformTemplate;
+
+/**
+ * {@link Platform} implementation for DuckDB.
+ *
+ * The {@code configName} {@code "duckdb"} makes Wayang resolve every property
+ * with the {@code wayang.duckdb.*} prefix.
+ */
+public class DuckDBPlatform extends JdbcPlatformTemplate {
+
+ private static final String PLATFORM_NAME = "DuckDB";
+
+ private static final String CONFIG_NAME = "duckdb";
+
+ private static DuckDBPlatform instance = null;
+
+ public static DuckDBPlatform getInstance() {
+ if (instance == null) {
+ instance = new DuckDBPlatform();
+ }
+ return instance;
+ }
+
+ protected DuckDBPlatform() {
+ super(PLATFORM_NAME, CONFIG_NAME);
+ }
+
+ @Override
+ public String getJdbcDriverClassName() {
+ return "org.duckdb.DuckDBDriver";
+ }
+
+}
diff --git a/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/plugin/DuckDBConversionsPlugin.java b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/plugin/DuckDBConversionsPlugin.java
new file mode 100644
index 000000000..aa797fd71
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/plugin/DuckDBConversionsPlugin.java
@@ -0,0 +1,59 @@
+/*
+ * 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.duckdb.plugin;
+
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.mapping.Mapping;
+import org.apache.wayang.core.optimizer.channels.ChannelConversion;
+import org.apache.wayang.core.platform.Platform;
+import org.apache.wayang.core.plugin.Plugin;
+import org.apache.wayang.java.platform.JavaPlatform;
+import org.apache.wayang.duckdb.channels.ChannelConversions;
+import org.apache.wayang.duckdb.platform.DuckDBPlatform;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * Provides only the {@link ChannelConversion}s for the {@link DuckDBPlatform}
+ * (no operator {@link Mapping}s), used to make DuckDB results consumable by
+ * other platforms without enabling operator pushdown.
+ */
+public class DuckDBConversionsPlugin implements Plugin {
+
+ @Override
+ public Collection getRequiredPlatforms() {
+ return Arrays.asList(DuckDBPlatform.getInstance(), JavaPlatform.getInstance());
+ }
+
+ @Override
+ public Collection getMappings() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Collection getChannelConversions() {
+ return ChannelConversions.ALL;
+ }
+
+ @Override
+ public void setProperties(Configuration configuration) {
+ }
+}
diff --git a/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/plugin/DuckDBPlugin.java b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/plugin/DuckDBPlugin.java
new file mode 100644
index 000000000..aaa6a0fa4
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/main/java/org/apache/wayang/duckdb/plugin/DuckDBPlugin.java
@@ -0,0 +1,58 @@
+/*
+ * 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.duckdb.plugin;
+
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.mapping.Mapping;
+import org.apache.wayang.core.optimizer.channels.ChannelConversion;
+import org.apache.wayang.core.plan.wayangplan.Operator;
+import org.apache.wayang.core.platform.Platform;
+import org.apache.wayang.core.plugin.Plugin;
+import org.apache.wayang.java.platform.JavaPlatform;
+import org.apache.wayang.duckdb.channels.ChannelConversions;
+import org.apache.wayang.duckdb.mapping.Mappings;
+import org.apache.wayang.duckdb.platform.DuckDBPlatform;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+/**
+ * Enables Wayang {@link Operator}s to be pushed down onto the {@link DuckDBPlatform}.
+ */
+public class DuckDBPlugin implements Plugin {
+
+ @Override
+ public Collection getRequiredPlatforms() {
+ return Arrays.asList(DuckDBPlatform.getInstance(), JavaPlatform.getInstance());
+ }
+
+ @Override
+ public Collection getMappings() {
+ return Mappings.ALL;
+ }
+
+ @Override
+ public Collection getChannelConversions() {
+ return ChannelConversions.ALL;
+ }
+
+ @Override
+ public void setProperties(Configuration configuration) {
+ }
+}
diff --git a/wayang-platforms/wayang-duckdb/src/main/resources/wayang-duckdb-defaults.properties b/wayang-platforms/wayang-duckdb/src/main/resources/wayang-duckdb-defaults.properties
new file mode 100644
index 000000000..8aec8f440
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/main/resources/wayang-duckdb-defaults.properties
@@ -0,0 +1,223 @@
+#
+# 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.
+#
+
+# Connection (override per deployment in wayang.properties).
+# Use jdbc:duckdb: for a persistent database file.
+wayang.duckdb.jdbc.url = jdbc:duckdb:
+wayang.duckdb.jdbc.user =
+wayang.duckdb.jdbc.password =
+wayang.duckdb.jdbc.driverName = org.duckdb.DuckDBDriver
+
+# ParquetSource support.
+# Use mappings to keep ParquetSource inputUrl as a canonical Parquet URI while
+# resolving it to a DuckDB-visible SQL relation during execution.
+# Format: parquet-uri=duckdb-relation;parquet-uri-2=duckdb-relation-2
+# wayang.duckdb.parquetsource.mappings = file:///data/orders.parquet=wayang_parquet.orders
+#
+# Optional auto-create support. When enabled, DuckDB creates a view over
+# read_parquet('${uri}') before composing the SQL stage.
+wayang.duckdb.parquetsource.auto-create = false
+wayang.duckdb.parquetsource.auto-create.relation-prefix = wayang_parquet_
+wayang.duckdb.parquetsource.auto-create.template = CREATE OR REPLACE VIEW ${relation} AS SELECT * FROM read_parquet('${uri}')
+#
+# Optional prepare SQL executed before DuckDB creates the Parquet view. Keep
+# statements idempotent: they can run during optimization cardinality estimation
+# and during execution. This can load DuckDB extensions such as httpfs.
+
+# Hardware profile used by LoadProfileToTimeConverter.
+wayang.duckdb.cpu.mhz = 2700
+wayang.duckdb.cores = 4
+wayang.duckdb.costs.fix = 0.0
+wayang.duckdb.costs.per-ms = 1.0
+
+# Reference cost model learned from the Trino Week8-style DuckDB pilot
+# (S01-S13, 10k/50k/100k/250k rows, 6 repetitions).
+# Re-profile for deployment-specific hardware, data shape, and PRAGMA threads.
+
+wayang.duckdb.tablesource.load.template = {\
+ "type":"mathex", "in":0, "out":1,\
+ "cpu":"?*out0 + ?",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.tablesource.load = {\
+ "type":"mathex",\
+ "in":0,\
+ "out":1,\
+ "cpu":"((10.472865061115924)*(out0))+(14042.766808200337)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
+}
+
+wayang.duckdb.parquetsource.load.template = {\
+ "type":"mathex", "in":0, "out":1,\
+ "cpu":"?*out0 + ?",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.parquetsource.load = {\
+ "type":"mathex",\
+ "in":0,\
+ "out":1,\
+ "cpu":"((10.472865061115924)*(out0))+(14042.766808200337)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
+}
+
+wayang.duckdb.filter.load.template = {\
+ "type":"mathex", "in":1, "out":1,\
+ "cpu":"?*in0 + ?",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.filter.load = {\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((55.53629275089796)*(in0))+(9844596.873788856)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
+}
+
+wayang.duckdb.projection.load.template = {\
+ "type":"mathex", "in":1, "out":1,\
+ "cpu":"?*in0 + ?",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.projection.load = {\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((24.619202590290122)*(in0))+(2184.9837993317874)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
+}
+
+wayang.duckdb.join.load.template = {\
+ "type":"mathex", "in":2, "out":1,\
+ "cpu":"?*in0 + ?*in1 + ?",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.join.load = {\
+ "type":"mathex",\
+ "in":2,\
+ "out":1,\
+ "cpu":"(((101.30552469545577)*(in0))+((30.896968724733654)*(in1)))+(53.481690928306364)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
+}
+
+wayang.duckdb.globalreduce.load.template = {\
+ "type":"mathex", "in":1, "out":1,\
+ "cpu":"?*in0 + ?",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.globalreduce.load = {\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((70.40718624423555)*(in0))+(8101.054898948163)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
+}
+
+wayang.duckdb.reduceby.load.template = {\
+ "type":"mathex", "in":1, "out":1,\
+ "cpu":"?*in0 + ?",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.reduceby.load = {\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((1.836076012140386)*(in0))+(20.74663077429906)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
+}
+
+wayang.duckdb.sort.load.template = {\
+ "type":"mathex", "in":1, "out":1,\
+ "cpu":"?*in0 + ?",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.sort.load = {\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((2622.8815961367363)*(in0))+(2537.744683180438)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
+}
+
+wayang.duckdb.tablesink.load.template = {\
+ "type":"mathex", "in":1, "out":0,\
+ "cpu":"?*in0 + ?",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.tablesink.load = {\
+ "type":"mathex",\
+ "in":1,\
+ "out":0,\
+ "cpu":"((2240.3138798129726)*(in0))+(0.7832474099914776)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
+}
+
+wayang.duckdb.sqltostream.load.query.template = {\
+ "type":"mathex", "in":1, "out":1,\
+ "cpu":"?*out0 + ?"\
+}
+wayang.duckdb.sqltostream.load.query = {\
+ "in":1, "out":1,\
+ "cpu":"${20*out0 + 200000}",\
+ "ram":"0",\
+ "p":0.9\
+}
+wayang.duckdb.sqltostream.load.output.template = {\
+ "type":"mathex", "in":1, "out":1,\
+ "cpu":"?*out0"\
+}
+wayang.duckdb.sqltostream.load.output = {\
+ "in":1, "out":1,\
+ "cpu":"${20*out0}",\
+ "ram":"0",\
+ "p":0.9\
+}
diff --git a/wayang-platforms/wayang-duckdb/src/test/java/org/apache/wayang/duckdb/DuckDBCostPilotIT.java b/wayang-platforms/wayang-duckdb/src/test/java/org/apache/wayang/duckdb/DuckDBCostPilotIT.java
new file mode 100644
index 000000000..e9f7f8868
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/test/java/org/apache/wayang/duckdb/DuckDBCostPilotIT.java
@@ -0,0 +1,777 @@
+/*
+ * 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.duckdb;
+
+import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.data.Tuple2;
+import org.apache.wayang.basic.function.ProjectionDescriptor;
+import org.apache.wayang.basic.operators.FilterOperator;
+import org.apache.wayang.basic.operators.GlobalReduceOperator;
+import org.apache.wayang.basic.operators.JoinOperator;
+import org.apache.wayang.basic.operators.MapOperator;
+import org.apache.wayang.basic.operators.ReduceByOperator;
+import org.apache.wayang.basic.operators.SortOperator;
+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.FunctionDescriptor;
+import org.apache.wayang.core.function.PredicateDescriptor;
+import org.apache.wayang.core.function.ReduceDescriptor;
+import org.apache.wayang.core.function.TransformationDescriptor;
+import org.apache.wayang.core.mapping.Mapping;
+import org.apache.wayang.core.mapping.OperatorPattern;
+import org.apache.wayang.core.mapping.PlanTransformation;
+import org.apache.wayang.core.mapping.ReplacementSubplanFactory;
+import org.apache.wayang.core.mapping.SubplanPattern;
+import org.apache.wayang.core.plan.wayangplan.WayangPlan;
+import org.apache.wayang.core.types.DataSetType;
+import org.apache.wayang.core.types.DataUnitType;
+import org.apache.wayang.duckdb.operators.DuckDBProjectionOperator;
+import org.apache.wayang.duckdb.operators.DuckDBTableSource;
+import org.apache.wayang.duckdb.platform.DuckDBPlatform;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.BufferedWriter;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Small DuckDB cost-profiling pilot.
+ */
+class DuckDBCostPilotIT {
+
+ private static final String SCHEMA = "wayang_profile";
+ private static final String CUSTOMERS_1K = SCHEMA + ".customers_1k";
+ private static final int[] ROW_COUNTS = parseIntList(System.getProperty(
+ "duckdb.profile.rowCounts",
+ "10000,50000,100000,250000"
+ ));
+ private static final String[] COLUMNS = {"order_id", "customer_id", "region", "amount", "bucket"};
+ private static final String[] JOIN_COLUMNS = {
+ "order_id", "customer_id", "region", "amount", "bucket", "cust_id", "tier"
+ };
+ private static final String[] JOIN_ORDER_TIER_AMOUNT_COLUMNS = {"order_id", "tier", "amount"};
+ private static final String[] JOIN_TIER_AMOUNT_COLUMNS = {"tier", "amount"};
+ private static final String JOIN_FLATTEN_NAME = "DuckDB profile join flatten";
+ private static final String JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME = "DuckDB profile join order tier amount flatten";
+ private static final String JOIN_TIER_AMOUNT_FLATTEN_NAME = "DuckDB profile join tier amount flatten";
+ private static final Path OUTPUT_DIR = Paths.get(System.getProperty(
+ "duckdb.profile.outputDir",
+ "target/cost-profiling/duckdb"
+ ));
+ private static final Path EXECUTIONS_PATH = OUTPUT_DIR.resolve("executions.json");
+ private static final Path CARDINALITIES_PATH = OUTPUT_DIR.resolve("cardinalities.json");
+ private static final Path MANIFEST_PATH = OUTPUT_DIR.resolve("manifest.csv");
+ private static final List PLAN_IDS = Arrays.asList(
+ System.getProperty(
+ "duckdb.profile.plans",
+ "S01,S02,S03,S04,S05,S06,S07,S08,S09,S10,S11,S12,S13"
+ ).split(",")
+ );
+ private static final int REPETITIONS = Integer.parseInt(
+ System.getProperty("duckdb.profile.repetitions", "6")
+ );
+ private static final boolean RESET_OUTPUT = Boolean.parseBoolean(
+ System.getProperty("duckdb.profile.reset", "true")
+ );
+
+ private static Path databaseFile;
+ private static String jdbcUrl;
+
+ @AfterEach
+ void cleanUpDatabaseFile() throws Exception {
+ if (databaseFile != null) {
+ Files.deleteIfExists(databaseFile);
+ Files.deleteIfExists(databaseFile.resolveSibling(databaseFile.getFileName() + ".wal"));
+ databaseFile = null;
+ }
+ }
+
+ @Test
+ void runPilot() throws Exception {
+ openFreshDatabase();
+ Files.createDirectories(OUTPUT_DIR);
+ initializeOutputFiles();
+
+ prepareTables();
+
+ for (int rowCount : ROW_COUNTS) {
+ for (String planId : PLAN_IDS) {
+ String normalizedPlanId = planId.trim();
+ runPlan(
+ normalizedPlanId,
+ getOperatorChain(normalizedPlanId),
+ rowCount,
+ getExpectedRows(normalizedPlanId, rowCount)
+ );
+ }
+ }
+
+ assertTrue(Files.exists(EXECUTIONS_PATH), "execution log should be written");
+ if (Files.exists(CARDINALITIES_PATH)) {
+ assertFalse(Files.readAllLines(CARDINALITIES_PATH).isEmpty(), "cardinality log should contain rows");
+ }
+ assertTrue(Files.exists(MANIFEST_PATH), "manifest should be written");
+ assertFalse(Files.readAllLines(MANIFEST_PATH).isEmpty(), "manifest should contain rows");
+ }
+
+ private void runPlan(String planId, String operatorChain, int rowCount, long expectedRows) throws Exception {
+ for (int repetition = 0; repetition < REPETITIONS; repetition++) {
+ boolean isWarmup = repetition == 0;
+ String runId = String.format("%s_%s_r%02d", planId, formatRows(rowCount), repetition);
+ String sourceTable = SCHEMA + ".orders_" + formatRows(rowCount);
+ String sinkTable = SCHEMA + ".sink_" + runId.toLowerCase();
+
+ dropTable(sinkTable);
+ WayangPlan plan = createPlan(planId, sourceTable, sinkTable);
+ wayangContext().execute(runId, plan);
+
+ long actualRows = queryLong("SELECT count(*) FROM " + sinkTable);
+ assertEquals(expectedRows, actualRows, runId + " row count");
+ appendManifest(runId, planId, operatorChain, rowCount, expectedRows, repetition, isWarmup, sinkTable);
+ dropTable(sinkTable);
+ }
+ }
+
+ private WayangPlan createPlan(String planId, String sourceTable, String sinkTable) {
+ DuckDBTableSource source = new DuckDBTableSource(sourceTable, COLUMNS);
+
+ if ("S01".equals(planId)) {
+ TableSink sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS);
+ source.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S02".equals(planId)) {
+ TableSink sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS);
+ FilterOperator filter = createAmerFilter();
+ source.connectTo(0, filter, 0);
+ filter.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S03".equals(planId)) {
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, "order_id", "amount");
+ MapOperator projection = createOrderAmountProjection();
+ source.connectTo(0, projection, 0);
+ projection.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S04".equals(planId)) {
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, "order_id", "amount");
+ FilterOperator filter = createAmerFilter();
+ MapOperator projection = createOrderAmountProjection();
+ source.connectTo(0, filter, 0);
+ filter.connectTo(0, projection, 0);
+ projection.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S05".equals(planId)) {
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, "total_amount");
+ GlobalReduceOperator reduce = createGlobalAmountReduceOperator();
+ source.connectTo(0, reduce, 0);
+ reduce.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S06".equals(planId)) {
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, "bucket", "total_amount");
+ ReduceByOperator reduceBy = createBucketReduceByOperator();
+ source.connectTo(0, reduceBy, 0);
+ reduceBy.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S07".equals(planId)) {
+ TableSink sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS);
+ SortOperator sort = createAmountSortOperator(3);
+ source.connectTo(0, sort, 0);
+ sort.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S08".equals(planId)) {
+ DuckDBTableSource customers = new DuckDBTableSource(CUSTOMERS_1K, "cust_id", "tier");
+ JoinOperator join = createCustomerJoinOperator(sourceTable);
+ MapOperator, Record> flatten = createJoinFlattenOperator();
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, JOIN_COLUMNS);
+ source.connectTo(0, join, 0);
+ customers.connectTo(0, join, 1);
+ join.connectTo(0, flatten, 0);
+ flatten.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S09".equals(planId)) {
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, "total_amount");
+ FilterOperator filter = createAmerFilter();
+ GlobalReduceOperator reduce = createGlobalAmountReduceOperator();
+ source.connectTo(0, filter, 0);
+ filter.connectTo(0, reduce, 0);
+ reduce.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S10".equals(planId)) {
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, "bucket", "total_amount");
+ FilterOperator filter = createAmerFilter();
+ ReduceByOperator reduceBy = createBucketReduceByOperator();
+ source.connectTo(0, filter, 0);
+ filter.connectTo(0, reduceBy, 0);
+ reduceBy.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S11".equals(planId)) {
+ TableSink sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS);
+ FilterOperator filter = createAmerFilter();
+ SortOperator sort = createAmountSortOperator(3);
+ source.connectTo(0, filter, 0);
+ filter.connectTo(0, sort, 0);
+ sort.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S12".equals(planId)) {
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, "order_id", "amount");
+ MapOperator projection = createOrderAmountProjection();
+ SortOperator sort = createAmountSortOperator(1);
+ source.connectTo(0, projection, 0);
+ projection.connectTo(0, sort, 0);
+ sort.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S13".equals(planId)) {
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, "order_id", "amount");
+ FilterOperator filter = createAmerFilter();
+ MapOperator projection = createOrderAmountProjection();
+ SortOperator sort = createAmountSortOperator(1);
+ source.connectTo(0, filter, 0);
+ filter.connectTo(0, projection, 0);
+ projection.connectTo(0, sort, 0);
+ sort.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S14".equals(planId)) {
+ DuckDBTableSource customers = new DuckDBTableSource(CUSTOMERS_1K, "cust_id", "tier");
+ FilterOperator filter = createAmerFilter();
+ JoinOperator join = createCustomerJoinOperator(sourceTable);
+ MapOperator, Record> flatten = createJoinFlattenOperator();
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, JOIN_COLUMNS);
+ source.connectTo(0, filter, 0);
+ filter.connectTo(0, join, 0);
+ customers.connectTo(0, join, 1);
+ join.connectTo(0, flatten, 0);
+ flatten.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S15".equals(planId)) {
+ DuckDBTableSource customers = new DuckDBTableSource(CUSTOMERS_1K, "cust_id", "tier");
+ JoinOperator join = createCustomerJoinOperator(sourceTable);
+ MapOperator, Record> flatten = createJoinOrderTierAmountFlattenOperator();
+ SortOperator sort = createAmountSortOperator(2);
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, JOIN_ORDER_TIER_AMOUNT_COLUMNS);
+ source.connectTo(0, join, 0);
+ customers.connectTo(0, join, 1);
+ join.connectTo(0, flatten, 0);
+ flatten.connectTo(0, sort, 0);
+ sort.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ if ("S16".equals(planId)) {
+ DuckDBTableSource customers = new DuckDBTableSource(CUSTOMERS_1K, "cust_id", "tier");
+ JoinOperator join = createCustomerJoinOperator(sourceTable);
+ MapOperator, Record> flatten = createJoinTierAmountFlattenOperator();
+ ReduceByOperator reduceBy = createTierReduceByOperator();
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", sinkTable, "tier", "total_amount");
+ source.connectTo(0, join, 0);
+ customers.connectTo(0, join, 1);
+ join.connectTo(0, flatten, 0);
+ flatten.connectTo(0, reduceBy, 0);
+ reduceBy.connectTo(0, sink, 0);
+ return new WayangPlan(sink);
+ }
+
+ throw new IllegalArgumentException("Unsupported pilot plan: " + planId);
+ }
+
+ private static GlobalReduceOperator createGlobalAmountReduceOperator() {
+ return new GlobalReduceOperator<>(
+ new ReduceDescriptor<>((left, right) -> left, Record.class)
+ .withSqlImplementation("SUM(amount) AS total_amount"),
+ DataSetType.createDefault(Record.class));
+ }
+
+ private static ReduceByOperator createBucketReduceByOperator() {
+ return new ReduceByOperator<>(
+ new TransformationDescriptor<>(
+ record -> new Record(record.getField(4)),
+ Record.class,
+ Record.class
+ ).withSqlImplementation("bucket", "bucket"),
+ new ReduceDescriptor<>((left, right) -> left, Record.class)
+ .withSqlImplementation("SUM(amount) AS total_amount"),
+ DataSetType.createDefault(Record.class));
+ }
+
+ private static ReduceByOperator createTierReduceByOperator() {
+ return new ReduceByOperator<>(
+ new TransformationDescriptor<>(
+ record -> new Record(record.getField(0)),
+ Record.class,
+ Record.class
+ ).withSqlImplementation("tier", "tier"),
+ new ReduceDescriptor<>((left, right) -> left, Record.class)
+ .withSqlImplementation("SUM(amount) AS total_amount"),
+ DataSetType.createDefault(Record.class));
+ }
+
+ private static SortOperator createAmountSortOperator(int amountFieldIndex) {
+ return new SortOperator<>(
+ new TransformationDescriptor<>(
+ record -> new Record(record.getField(amountFieldIndex)),
+ Record.class,
+ Record.class
+ ).withSqlImplementation("amount", "ASC"),
+ DataSetType.createDefault(Record.class));
+ }
+
+ private static JoinOperator createCustomerJoinOperator(String sourceTable) {
+ return new JoinOperator<>(
+ new TransformationDescriptor<>(
+ record -> new Record(record.getField(1)),
+ Record.class,
+ Record.class
+ ).withSqlImplementation(sourceTable, "customer_id"),
+ new TransformationDescriptor<>(
+ record -> new Record(record.getField(0)),
+ Record.class,
+ Record.class
+ ).withSqlImplementation(CUSTOMERS_1K, "cust_id"));
+ }
+
+ private static FilterOperator createAmerFilter() {
+ return new FilterOperator<>(
+ new PredicateDescriptor<>(
+ (Record record) -> "AMER".equals(record.getField(2)),
+ Record.class
+ ).withSqlImplementation("region = 'AMER'")
+ );
+ }
+
+ private static MapOperator createOrderAmountProjection() {
+ return new MapOperator<>(
+ ProjectionDescriptor.createForRecords(
+ new RecordType(COLUMNS),
+ "order_id", "amount"),
+ DataSetType.createDefault(Record.class),
+ DataSetType.createDefault(Record.class));
+ }
+
+ private static MapOperator, Record> createJoinFlattenOperator() {
+ return createJoinFlattenOperator(new JoinFlattenFunction(), JOIN_FLATTEN_NAME);
+ }
+
+ private static MapOperator, Record> createJoinOrderTierAmountFlattenOperator() {
+ return createJoinFlattenOperator(new JoinOrderTierAmountFlattenFunction(), JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME);
+ }
+
+ private static MapOperator, Record> createJoinTierAmountFlattenOperator() {
+ return createJoinFlattenOperator(new JoinTierAmountFlattenFunction(), JOIN_TIER_AMOUNT_FLATTEN_NAME);
+ }
+
+ private static MapOperator, Record> createJoinFlattenOperator(
+ FunctionDescriptor.SerializableFunction, Record> function,
+ String name) {
+ MapOperator, Record> operator = new MapOperator<>(
+ new TransformationDescriptor<>(
+ function,
+ DataUnitType.createBasicUnchecked(Tuple2.class),
+ DataUnitType.createBasic(Record.class)),
+ DataSetType.createDefaultUnchecked(Tuple2.class),
+ DataSetType.createDefault(Record.class));
+ operator.setName(name);
+ return operator;
+ }
+
+ private WayangContext wayangContext() {
+ Configuration configuration = new Configuration();
+ configuration.setProperty("wayang.duckdb.jdbc.url", jdbcUrl);
+ configuration.setProperty("wayang.duckdb.jdbc.user", "");
+ configuration.setProperty("wayang.duckdb.jdbc.password", "");
+ configuration.setProperty("wayang.core.log.enabled", "true");
+ configuration.setProperty("wayang.core.explain.enabled", "false");
+ configuration.setProperty("wayang.core.log.executions", EXECUTIONS_PATH.toString().replace('\\', '/'));
+ configuration.setProperty("wayang.core.log.cardinalities", CARDINALITIES_PATH.toString().replace('\\', '/'));
+ configuration.getMappingProvider().addAllToWhitelist(
+ Collections.singleton(new JoinFlattenMapping()));
+ return new WayangContext(configuration).withPlugin(DuckDB.plugin());
+ }
+
+ private static void prepareTables() throws Exception {
+ try (Connection connection = jdbc(); Statement statement = connection.createStatement()) {
+ statement.execute("CREATE SCHEMA IF NOT EXISTS " + SCHEMA);
+ for (int rowCount : ROW_COUNTS) {
+ String table = SCHEMA + ".orders_" + formatRows(rowCount);
+ statement.execute("DROP TABLE IF EXISTS " + table);
+ statement.execute("CREATE TABLE " + table + " AS "
+ + "SELECT "
+ + "CAST(n AS BIGINT) AS order_id, "
+ + "CAST(n % 1000 AS BIGINT) AS customer_id, "
+ + "CASE WHEN n % 2 = 0 THEN 'AMER' ELSE 'EMEA' END AS region, "
+ + "CAST(n % 10000 AS DOUBLE) AS amount, "
+ + "CAST(n % 100 AS BIGINT) AS bucket "
+ + "FROM range(1, " + (rowCount + 1) + ") AS t(n)");
+ assertEquals(rowCount, queryLong("SELECT count(*) FROM " + table), table + " row count");
+ assertEquals(rowCount / 2, queryLong("SELECT count(*) FROM " + table + " WHERE region = 'AMER'"),
+ table + " AMER row count");
+ }
+ statement.execute("DROP TABLE IF EXISTS " + CUSTOMERS_1K);
+ statement.execute("CREATE TABLE " + CUSTOMERS_1K + " AS "
+ + "SELECT "
+ + "CAST(n - 1 AS BIGINT) AS cust_id, "
+ + "CASE WHEN n % 2 = 0 THEN 'GOLD' ELSE 'SILVER' END AS tier "
+ + "FROM range(1, 1001) AS t(n)");
+ assertEquals(1000, queryLong("SELECT count(*) FROM " + CUSTOMERS_1K), CUSTOMERS_1K + " row count");
+ }
+ }
+
+ private static String formatRows(int rowCount) {
+ if (rowCount % 1000 == 0) {
+ return (rowCount / 1000) + "k";
+ }
+ return String.valueOf(rowCount);
+ }
+
+ private static void initializeOutputFiles() throws Exception {
+ if (RESET_OUTPUT) {
+ Files.deleteIfExists(EXECUTIONS_PATH);
+ Files.deleteIfExists(CARDINALITIES_PATH);
+ writeManifestHeader();
+ } else if (!Files.exists(MANIFEST_PATH)) {
+ writeManifestHeader();
+ }
+ }
+
+ private static void writeManifestHeader() throws Exception {
+ try (BufferedWriter writer = Files.newBufferedWriter(MANIFEST_PATH, StandardCharsets.UTF_8)) {
+ writer.write("run_id,plan_id,operator_chain,input_rows_left,input_rows_right,expected_output_rows,"
+ + "selectivity,repetition,is_warmup,sink_table,status,notes");
+ writer.newLine();
+ }
+ }
+
+ private static void appendManifest(
+ String runId,
+ String planId,
+ String operatorChain,
+ int inputRows,
+ long expectedOutputRows,
+ int repetition,
+ boolean isWarmup,
+ String sinkTable) throws Exception {
+ try (BufferedWriter writer = Files.newBufferedWriter(
+ MANIFEST_PATH,
+ StandardCharsets.UTF_8,
+ java.nio.file.StandardOpenOption.APPEND)) {
+ writer.write(String.join(",",
+ runId,
+ planId,
+ operatorChain,
+ String.valueOf(inputRows),
+ hasJoin(planId) ? "1000" : "",
+ String.valueOf(expectedOutputRows),
+ hasFilter(planId) ? "0.5" : "1.0",
+ String.valueOf(repetition),
+ String.valueOf(isWarmup),
+ sinkTable,
+ "ok",
+ ""));
+ writer.newLine();
+ }
+ }
+
+ private static String getOperatorChain(String planId) {
+ switch (planId) {
+ case "S01":
+ return "TableSource->TableSink";
+ case "S02":
+ return "TableSource->Filter(50%)->TableSink";
+ case "S03":
+ return "TableSource->Projection->TableSink";
+ case "S04":
+ return "TableSource->Filter(50%)->Projection->TableSink";
+ case "S05":
+ return "TableSource->GlobalReduce->TableSink";
+ case "S06":
+ return "TableSource->ReduceBy(bucket)->TableSink";
+ case "S07":
+ return "TableSource->Sort(amount)->TableSink";
+ case "S08":
+ return "Orders->Join(Customers 1k)->Projection->TableSink";
+ case "S09":
+ return "TableSource->Filter(50%)->GlobalReduce->TableSink";
+ case "S10":
+ return "TableSource->Filter(50%)->ReduceBy(bucket)->TableSink";
+ case "S11":
+ return "TableSource->Filter(50%)->Sort(amount)->TableSink";
+ case "S12":
+ return "TableSource->Projection(order_id,amount)->Sort(amount)->TableSink";
+ case "S13":
+ return "TableSource->Filter(50%)->Projection(order_id,amount)->Sort(amount)->TableSink";
+ case "S14":
+ return "Orders->Filter(50%)->Join(Customers 1k)->Projection->TableSink";
+ case "S15":
+ return "Orders->Join(Customers 1k)->Projection(order_id,tier,amount)->Sort(amount)->TableSink";
+ case "S16":
+ return "Orders->Join(Customers 1k)->Projection(tier,amount)->ReduceBy(tier)->TableSink";
+ default:
+ throw new IllegalArgumentException("Unsupported pilot plan: " + planId);
+ }
+ }
+
+ private static long getExpectedRows(String planId, int rowCount) {
+ if ("S05".equals(planId) || "S09".equals(planId)) {
+ return 1;
+ }
+ if ("S06".equals(planId)) {
+ return 100;
+ }
+ if ("S10".equals(planId)) {
+ return 50;
+ }
+ if ("S16".equals(planId)) {
+ return 2;
+ }
+ return hasFilter(planId) ? rowCount / 2 : rowCount;
+ }
+
+ private static boolean hasFilter(String planId) {
+ return "S02".equals(planId)
+ || "S04".equals(planId)
+ || "S09".equals(planId)
+ || "S10".equals(planId)
+ || "S11".equals(planId)
+ || "S13".equals(planId)
+ || "S14".equals(planId);
+ }
+
+ private static boolean hasJoin(String planId) {
+ return "S08".equals(planId)
+ || "S14".equals(planId)
+ || "S15".equals(planId)
+ || "S16".equals(planId);
+ }
+
+ private static int[] parseIntList(String value) {
+ return Arrays.stream(value.split(","))
+ .map(String::trim)
+ .filter(token -> !token.isEmpty())
+ .mapToInt(Integer::parseInt)
+ .toArray();
+ }
+
+ private static long queryLong(String sql) throws Exception {
+ try (Connection connection = jdbc();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ resultSet.next();
+ return resultSet.getLong(1);
+ }
+ }
+
+ private static void dropTable(String table) throws Exception {
+ try (Connection connection = jdbc(); Statement statement = connection.createStatement()) {
+ statement.execute("DROP TABLE IF EXISTS " + table);
+ }
+ }
+
+ private static Connection jdbc() throws Exception {
+ return DriverManager.getConnection(jdbcUrl);
+ }
+
+ private static void openFreshDatabase() throws Exception {
+ databaseFile = Files.createTempFile("wayang-duckdb-cost-", ".duckdb");
+ Files.deleteIfExists(databaseFile);
+ jdbcUrl = "jdbc:duckdb:" + databaseFile.toAbsolutePath();
+ }
+
+ private static Record flattenJoinResult(Object joinResult) {
+ if (joinResult instanceof Record) {
+ return (Record) joinResult;
+ }
+ Tuple2, ?> pair = (Tuple2, ?>) joinResult;
+ Record left = (Record) pair.field0;
+ Record right = (Record) pair.field1;
+ return new Record(
+ left.getField(0),
+ left.getField(1),
+ left.getField(2),
+ left.getField(3),
+ left.getField(4),
+ right.getField(0),
+ right.getField(1));
+ }
+
+ private static Record flattenJoinOrderTierAmountResult(Object joinResult) {
+ if (joinResult instanceof Record) {
+ Record record = (Record) joinResult;
+ return new Record(record.getField(0), record.getField(6), record.getField(3));
+ }
+ Tuple2, ?> pair = (Tuple2, ?>) joinResult;
+ Record left = (Record) pair.field0;
+ Record right = (Record) pair.field1;
+ return new Record(left.getField(0), right.getField(1), left.getField(3));
+ }
+
+ private static Record flattenJoinTierAmountResult(Object joinResult) {
+ if (joinResult instanceof Record) {
+ Record record = (Record) joinResult;
+ return new Record(record.getField(6), record.getField(3));
+ }
+ Tuple2, ?> pair = (Tuple2, ?>) joinResult;
+ Record left = (Record) pair.field0;
+ Record right = (Record) pair.field1;
+ return new Record(right.getField(1), left.getField(3));
+ }
+
+ private static final class JoinFlattenFunction implements
+ FunctionDescriptor.SerializableFunction, Record> {
+
+ @Override
+ public Record apply(Tuple2 tuple) {
+ return flattenJoinResult(tuple);
+ }
+ }
+
+ private static final class JoinOrderTierAmountFlattenFunction implements
+ FunctionDescriptor.SerializableFunction, Record> {
+
+ @Override
+ public Record apply(Tuple2 tuple) {
+ return flattenJoinOrderTierAmountResult(tuple);
+ }
+ }
+
+ private static final class JoinTierAmountFlattenFunction implements
+ FunctionDescriptor.SerializableFunction, Record> {
+
+ @Override
+ public Record apply(Tuple2 tuple) {
+ return flattenJoinTierAmountResult(tuple);
+ }
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static final class JoinFlattenMapping implements Mapping {
+
+ @Override
+ public java.util.Collection getTransformations() {
+ OperatorPattern pattern = new OperatorPattern(
+ "joinFlatten",
+ new MapOperator(null, DataSetType.none(), DataSetType.createDefault(Record.class)),
+ false)
+ .withAdditionalTest(operator -> isJoinFlattenName(((MapOperator) operator).getName()));
+
+ ReplacementSubplanFactory factory = new ReplacementSubplanFactory.OfSingleOperators(
+ (matchedOperator, epoch) -> createDuckDBProjection(matchedOperator.getName()).at(epoch));
+
+ return Collections.singleton(new PlanTransformation(
+ SubplanPattern.createSingleton(pattern),
+ factory,
+ DuckDBPlatform.getInstance()));
+ }
+
+ private static DuckDBProjectionOperator createDuckDBProjection(String operatorName) {
+ ProjectionDescriptor, Record> descriptor = new ProjectionDescriptor<>(
+ getJoinFlattenFunction(operatorName),
+ Arrays.asList(getJoinFlattenColumns(operatorName)),
+ DataUnitType.createBasicUnchecked(Tuple2.class),
+ DataUnitType.createBasic(Record.class));
+ MapOperator, Record> projection = new MapOperator<>(
+ descriptor,
+ DataSetType.createDefaultUnchecked(Tuple2.class),
+ DataSetType.createDefault(Record.class));
+ projection.setName(operatorName);
+ return new DuckDBProjectionOperator((MapOperator) (MapOperator) projection);
+ }
+
+ private static boolean isJoinFlattenName(String operatorName) {
+ return JOIN_FLATTEN_NAME.equals(operatorName)
+ || JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName)
+ || JOIN_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName);
+ }
+
+ private static String[] getJoinFlattenColumns(String operatorName) {
+ if (JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName)) {
+ return JOIN_ORDER_TIER_AMOUNT_COLUMNS;
+ }
+ if (JOIN_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName)) {
+ return JOIN_TIER_AMOUNT_COLUMNS;
+ }
+ return JOIN_COLUMNS;
+ }
+
+ private static FunctionDescriptor.SerializableFunction, Record> getJoinFlattenFunction(
+ String operatorName) {
+ if (JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName)) {
+ return new JoinOrderTierAmountFlattenFunction();
+ }
+ if (JOIN_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName)) {
+ return new JoinTierAmountFlattenFunction();
+ }
+ return new JoinFlattenFunction();
+ }
+ }
+}
diff --git a/wayang-platforms/wayang-duckdb/src/test/java/org/apache/wayang/duckdb/DuckDBOperatorsIT.java b/wayang-platforms/wayang-duckdb/src/test/java/org/apache/wayang/duckdb/DuckDBOperatorsIT.java
new file mode 100644
index 000000000..9483fe0b7
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/test/java/org/apache/wayang/duckdb/DuckDBOperatorsIT.java
@@ -0,0 +1,636 @@
+/*
+ * 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.duckdb;
+
+import org.apache.wayang.api.DataQuantaBuilder;
+import org.apache.wayang.api.JavaPlanBuilder;
+import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.data.Tuple2;
+import org.apache.wayang.basic.function.ProjectionDescriptor;
+import org.apache.wayang.basic.operators.FilterOperator;
+import org.apache.wayang.basic.operators.GlobalReduceOperator;
+import org.apache.wayang.basic.operators.JoinOperator;
+import org.apache.wayang.basic.operators.MapOperator;
+import org.apache.wayang.basic.operators.ReduceByOperator;
+import org.apache.wayang.basic.operators.SortOperator;
+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.FunctionDescriptor;
+import org.apache.wayang.core.function.PredicateDescriptor;
+import org.apache.wayang.core.function.ReduceDescriptor;
+import org.apache.wayang.core.function.TransformationDescriptor;
+import org.apache.wayang.core.mapping.Mapping;
+import org.apache.wayang.core.mapping.OperatorPattern;
+import org.apache.wayang.core.mapping.PlanTransformation;
+import org.apache.wayang.core.mapping.ReplacementSubplanFactory;
+import org.apache.wayang.core.mapping.SubplanPattern;
+import org.apache.wayang.core.plan.wayangplan.WayangPlan;
+import org.apache.wayang.core.types.DataSetType;
+import org.apache.wayang.core.types.DataUnitType;
+import org.apache.wayang.duckdb.operators.DuckDBProjectionOperator;
+import org.apache.wayang.duckdb.operators.DuckDBTableSource;
+import org.apache.wayang.duckdb.platform.DuckDBPlatform;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Embedded end-to-end tests for every operator the DuckDB platform implements.
+ *
+ * Coverage mirrors {@code TrinoOperatorsIT}: {@code TableSource},
+ * {@code Filter}, {@code Projection}, {@code Join}, {@code GlobalReduce},
+ * {@code ReduceBy}, {@code Sort}, and {@code TableSink}, plus five
+ * JavaPlanBuilder combinations. Each Wayang plan registers only
+ * {@link DuckDB#plugin()} and ends in a DuckDB table sink, so no Java-side
+ * operator implementation is needed to compute the result.
+ */
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+class DuckDBOperatorsIT {
+
+ private static final String SCHEMA = "wayang_it";
+ private static final String ORDERS = SCHEMA + ".orders";
+ private static final String CUSTOMERS = SCHEMA + ".customers";
+ private static final String SINK_TABLE_NAME = "operator_result";
+ private static final String SINK_TABLE = SCHEMA + "." + SINK_TABLE_NAME;
+ private static final String[] JOIN_COLUMNS = {
+ "order_id", "customer_id", "region", "amount", "cust_id", "name", "tier"
+ };
+ private static final String JOIN_FLATTEN_NAME = "DuckDB test-only join flatten";
+
+ private static Path databaseFile;
+ private static String jdbcUrl;
+ private static boolean deleteDatabaseFile;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ jdbcUrl = System.getProperty(
+ "duckdb.url",
+ System.getenv().getOrDefault("DUCKDB_JDBC_URL", ""));
+ if (jdbcUrl.isEmpty()) {
+ databaseFile = Files.createTempFile("wayang-duckdb-operators-", ".duckdb");
+ Files.deleteIfExists(databaseFile);
+ jdbcUrl = "jdbc:duckdb:" + databaseFile.toAbsolutePath();
+ deleteDatabaseFile = true;
+ } else {
+ jdbcUrl = normalizeDuckDbUrl(jdbcUrl);
+ }
+
+ try (Connection connection = jdbc(); Statement statement = connection.createStatement()) {
+ statement.execute("CREATE SCHEMA IF NOT EXISTS " + SCHEMA);
+ statement.execute("DROP TABLE IF EXISTS " + SINK_TABLE);
+ statement.execute("DROP TABLE IF EXISTS " + ORDERS);
+ statement.execute("DROP TABLE IF EXISTS " + CUSTOMERS);
+ 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)");
+
+ statement.execute("CREATE TABLE " + CUSTOMERS + " ("
+ + "cust_id BIGINT, name VARCHAR, tier VARCHAR)");
+ statement.execute("INSERT INTO " + CUSTOMERS + " VALUES "
+ + "(100, 'Acme', 'GOLD'),"
+ + "(101, 'Globex', 'SILVER'),"
+ + "(102, 'Initech','BRONZE')");
+ }
+ }
+
+ @AfterAll
+ static void tearDown() throws Exception {
+ if (deleteDatabaseFile && databaseFile != null) {
+ Files.deleteIfExists(databaseFile);
+ Files.deleteIfExists(databaseFile.resolveSibling(databaseFile.getFileName() + ".wal"));
+ }
+ }
+
+ @Test
+ @Order(1)
+ void loadsDuckDbDriverAndRunsQuery() throws Exception {
+ Class.forName(DuckDBPlatform.getInstance().getJdbcDriverClassName());
+
+ try (Connection connection = DriverManager.getConnection("jdbc:duckdb:");
+ ResultSet resultSet = connection.createStatement().executeQuery("SELECT 1")) {
+ resultSet.next();
+ assertEquals(1, resultSet.getInt(1));
+ }
+ }
+
+ @Test
+ @Order(2)
+ void tableSource() {
+ DuckDBTableSource src = new DuckDBTableSource(ORDERS, "order_id", "customer_id", "region", "amount");
+ TableSink sink = tableSink("order_id", "customer_id", "region", "amount");
+ src.connectTo(0, sink, 0);
+
+ wayangContext().execute(new WayangPlan(sink));
+
+ assertEquals(6, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ }
+
+ @Test
+ @Order(3)
+ void filter() {
+ DuckDBTableSource src = new DuckDBTableSource(ORDERS, "order_id", "customer_id", "region", "amount");
+ FilterOperator filter = new FilterOperator<>(
+ new PredicateDescriptor<>(
+ (Record record) -> "AMER".equals(record.getField(2)), Record.class)
+ .withSqlImplementation("region = 'AMER'"));
+ TableSink sink = tableSink("order_id", "customer_id", "region", "amount");
+ src.connectTo(0, filter, 0);
+ filter.connectTo(0, sink, 0);
+
+ wayangContext().execute(new WayangPlan(sink));
+
+ assertEquals(3, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ assertEquals(0, queryLong("SELECT SUM(CASE WHEN region <> 'AMER' THEN 1 ELSE 0 END) FROM " + SINK_TABLE));
+ }
+
+ @Test
+ @Order(4)
+ void projection() {
+ DuckDBTableSource src = new DuckDBTableSource(ORDERS, "order_id", "customer_id", "region", "amount");
+ FilterOperator filter = new FilterOperator<>(
+ new PredicateDescriptor<>(
+ (Record record) -> "AMER".equals(record.getField(2)), Record.class)
+ .withSqlImplementation("region = 'AMER'"));
+ MapOperator projection = new MapOperator<>(
+ ProjectionDescriptor.createForRecords(
+ new RecordType("order_id", "customer_id", "region", "amount"),
+ "region", "amount"),
+ DataSetType.createDefault(Record.class),
+ DataSetType.createDefault(Record.class));
+ TableSink sink = tableSink("region", "amount");
+ src.connectTo(0, filter, 0);
+ filter.connectTo(0, projection, 0);
+ projection.connectTo(0, sink, 0);
+
+ wayangContext().execute(new WayangPlan(sink));
+
+ assertEquals(3, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ assertEquals(2, queryLong(
+ "SELECT count(*) FROM information_schema.columns "
+ + "WHERE table_schema = '" + SCHEMA + "' AND table_name = '" + SINK_TABLE_NAME + "'"));
+ }
+
+ @Test
+ @Order(5)
+ void join() {
+ DuckDBTableSource orders = new DuckDBTableSource(
+ ORDERS, "order_id", "customer_id", "region", "amount");
+ DuckDBTableSource customers = new DuckDBTableSource(
+ CUSTOMERS, "cust_id", "name", "tier");
+ JoinOperator join = new JoinOperator<>(
+ new TransformationDescriptor<>(
+ (Record record) -> new Record(record.getField(1)), Record.class, Record.class)
+ .withSqlImplementation(ORDERS, "customer_id"),
+ new TransformationDescriptor<>(
+ (Record record) -> new Record(record.getField(0)), Record.class, Record.class)
+ .withSqlImplementation(CUSTOMERS, "cust_id"));
+ MapOperator, Record> flatten = joinFlattenOperator();
+ TableSink sink = tableSink(JOIN_COLUMNS);
+ orders.connectTo(0, join, 0);
+ customers.connectTo(0, join, 1);
+ join.connectTo(0, flatten, 0);
+ flatten.connectTo(0, sink, 0);
+
+ wayangContext().execute(new WayangPlan(sink));
+
+ assertEquals(6, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ assertEquals(0, queryLong("SELECT SUM(CASE WHEN customer_id <> cust_id THEN 1 ELSE 0 END) FROM "
+ + SINK_TABLE));
+ }
+
+ @Test
+ @Order(6)
+ void globalReduce() {
+ DuckDBTableSource src = new DuckDBTableSource(ORDERS, "order_id", "customer_id", "region", "amount");
+ GlobalReduceOperator reduce = new GlobalReduceOperator<>(
+ new ReduceDescriptor<>((left, right) -> left, Record.class)
+ .withSqlImplementation("SUM(amount) AS total_amount"),
+ DataSetType.createDefault(Record.class));
+ TableSink sink = tableSink("total_amount");
+ src.connectTo(0, reduce, 0);
+ reduce.connectTo(0, sink, 0);
+
+ wayangContext().execute(new WayangPlan(sink));
+
+ assertSingleDoubleResult(7231.25);
+ }
+
+ @Test
+ @Order(7)
+ void reduceBy() {
+ DuckDBTableSource src = new DuckDBTableSource(ORDERS, "order_id", "customer_id", "region", "amount");
+ ReduceByOperator reduceBy = new ReduceByOperator<>(
+ new TransformationDescriptor<>(
+ (Record record) -> new Record(record.getField(2)), Record.class, Record.class)
+ .withSqlImplementation("region", "region"),
+ new ReduceDescriptor<>((left, right) -> left, Record.class)
+ .withSqlImplementation("SUM(amount) AS total_amount"),
+ DataSetType.createDefault(Record.class));
+ TableSink sink = tableSink("region", "total_amount");
+ src.connectTo(0, reduceBy, 0);
+ reduceBy.connectTo(0, sink, 0);
+
+ wayangContext().execute(new WayangPlan(sink));
+
+ Map sums = readRegionSums();
+ assertEquals(3, sums.size());
+ assertEquals(3830.75, sums.get("AMER"), 0.01);
+ assertEquals(1900.5, sums.get("EMEA"), 0.01);
+ assertEquals(1500.0, sums.get("APAC"), 0.01);
+ }
+
+ @Test
+ @Order(8)
+ void sort() {
+ DuckDBTableSource src = new DuckDBTableSource(ORDERS, "order_id", "customer_id", "region", "amount");
+ SortOperator sort = new SortOperator<>(
+ new TransformationDescriptor<>(
+ (Record record) -> new Record(record.getField(3)), Record.class, Record.class)
+ .withSqlImplementation("amount", "ASC"),
+ DataSetType.createDefault(Record.class));
+ TableSink sink = tableSink("order_id", "customer_id", "region", "amount");
+ src.connectTo(0, sort, 0);
+ sort.connectTo(0, sink, 0);
+
+ wayangContext().execute(new WayangPlan(sink));
+
+ assertEquals(6, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ assertEquals(680.5, queryDouble("SELECT min(amount) FROM " + SINK_TABLE), 0.001);
+ assertEquals(2200.0, queryDouble("SELECT max(amount) FROM " + SINK_TABLE), 0.001);
+ }
+
+ @Test
+ @Order(9)
+ void tableSink() {
+ DuckDBTableSource src = new DuckDBTableSource(ORDERS, "order_id", "customer_id", "region", "amount");
+ FilterOperator filter = new FilterOperator<>(
+ new PredicateDescriptor<>(
+ (Record record) -> "AMER".equals(record.getField(2)), Record.class)
+ .withSqlImplementation("region = 'AMER'"));
+ TableSink sink = new TableSink<>(
+ new Properties(), "overwrite", SINK_TABLE,
+ "order_id", "customer_id", "region", "amount");
+ src.connectTo(0, filter, 0);
+ filter.connectTo(0, sink, 0);
+
+ wayangContext().execute(new WayangPlan(sink));
+
+ assertEquals(3, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ assertEquals(0, queryLong("SELECT SUM(CASE WHEN region <> 'AMER' THEN 1 ELSE 0 END) FROM " + SINK_TABLE));
+ }
+
+ @Test
+ @Order(10)
+ void javaPlanBuilderReadTableFilterProjection() {
+ new JavaPlanBuilder(
+ wayangContext(), "DuckDB JavaPlanBuilder readTable integration test")
+ .readTable(new DuckDBTableSource(
+ ORDERS, "order_id", "customer_id", "region", "amount"))
+ .filter(record -> "AMER".equals(record.getField(2)))
+ .withSqlUdf("region = 'AMER'")
+ .asRecords()
+ .projectRecords(new String[]{"order_id", "amount"})
+ .writeTable(SINK_TABLE, "overwrite", new String[]{"order_id", "amount"}, new Properties());
+
+ assertEquals(3, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ assertEquals(0, queryLong(
+ "SELECT SUM(CASE WHEN amount NOT IN (2200.0, 680.5, 950.25) THEN 1 ELSE 0 END) FROM "
+ + SINK_TABLE));
+ }
+
+ @Test
+ @Order(11)
+ void javaPlanBuilderReadTableFilterGlobalReduce() {
+ new JavaPlanBuilder(
+ wayangContext(), "DuckDB JavaPlanBuilder global reduce integration test")
+ .readTable(new DuckDBTableSource(
+ ORDERS, "order_id", "customer_id", "region", "amount"))
+ .filter(record -> "AMER".equals(record.getField(2)))
+ .withSqlUdf("region = 'AMER'")
+ .reduce((left, right) -> left)
+ .withSqlUdf("SUM(amount) AS total_amount")
+ .writeTable(SINK_TABLE, "overwrite", new String[]{"total_amount"}, new Properties());
+
+ assertSingleDoubleResult(3830.75);
+ }
+
+ @Test
+ @Order(12)
+ void javaPlanBuilderReadTableReduceBySort() {
+ new JavaPlanBuilder(
+ wayangContext(), "DuckDB JavaPlanBuilder reduce-by and sort integration test")
+ .readTable(new DuckDBTableSource(
+ ORDERS, "order_id", "customer_id", "region", "amount"))
+ .reduceByKey(
+ record -> new Record(record.getField(2)),
+ (left, right) -> left)
+ .withSqlUdfs("region", "SUM(amount) AS total_amount")
+ .sort(record -> new Record(record.getField(0)))
+ .withSqlUdf("region", "ASC")
+ .writeTable(SINK_TABLE, "overwrite", new String[]{"region", "total_amount"}, new Properties());
+
+ assertEquals("AMER,APAC,EMEA", queryString(
+ "SELECT string_agg(region, ',' ORDER BY region) FROM " + SINK_TABLE));
+ }
+
+ @Test
+ @Order(13)
+ void javaPlanBuilderReadTableFilterProjectionTableSink() {
+ new JavaPlanBuilder(wayangContext(), "DuckDB JavaPlanBuilder table sink integration test")
+ .readTable(new DuckDBTableSource(
+ ORDERS, "order_id", "customer_id", "region", "amount"))
+ .filter(record -> "AMER".equals(record.getField(2)))
+ .withSqlUdf("region = 'AMER'")
+ .asRecords()
+ .projectRecords(new String[]{"order_id", "amount"})
+ .writeTable(
+ SINK_TABLE,
+ "overwrite",
+ new String[]{"order_id", "amount"},
+ new Properties());
+
+ assertEquals(3, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ assertEquals(2, queryLong(
+ "SELECT count(*) FROM information_schema.columns "
+ + "WHERE table_schema = '" + SCHEMA + "' AND table_name = '" + SINK_TABLE_NAME + "'"));
+ }
+
+ @Test
+ @Order(14)
+ void javaPlanBuilderReadTableJoin() {
+ JavaPlanBuilder plan = new JavaPlanBuilder(
+ wayangContext(), "DuckDB JavaPlanBuilder join integration test");
+ DataQuantaBuilder, Record> orders = plan.readTable(new DuckDBTableSource(
+ ORDERS, "order_id", "customer_id", "region", "amount"));
+ DataQuantaBuilder, Record> customers = plan.readTable(new DuckDBTableSource(
+ CUSTOMERS, "cust_id", "name", "tier"));
+
+ orders
+ .join(
+ record -> new Record(record.getField(1)),
+ customers,
+ record -> new Record(record.getField(0)))
+ .withSqlUdfs(ORDERS, "customer_id", CUSTOMERS, "cust_id")
+ .map(new JoinFlattenFunction())
+ .withName(JOIN_FLATTEN_NAME)
+ .writeTable(SINK_TABLE, "overwrite", JOIN_COLUMNS, new Properties());
+
+ assertEquals(6, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ assertEquals(0, queryLong("SELECT SUM(CASE WHEN customer_id <> cust_id THEN 1 ELSE 0 END) FROM "
+ + SINK_TABLE));
+ }
+
+ @Test
+ @Order(15)
+ void generatedSqlContainsPushdownShapes() {
+ String filterSql = captureStdout(this::filter);
+ assertTrue(filterSql.contains("WHERE region = 'AMER'"));
+
+ String projectionSql = captureStdout(this::projection);
+ assertTrue(projectionSql.contains("SELECT region, amount FROM " + ORDERS));
+ assertTrue(projectionSql.contains("WHERE region = 'AMER'"));
+
+ String joinSql = captureStdout(this::join);
+ assertTrue(joinSql.contains("JOIN " + CUSTOMERS));
+ assertTrue(joinSql.contains(CUSTOMERS + ".cust_id=" + ORDERS + ".customer_id"));
+
+ String reduceBySql = captureStdout(this::reduceBy);
+ assertTrue(reduceBySql.contains("GROUP BY region"));
+
+ String sortSql = captureStdout(this::sort);
+ assertTrue(sortSql.contains("ORDER BY amount ASC"));
+ }
+
+ private WayangContext wayangContext() {
+ Configuration config = new Configuration();
+ config.setProperty("wayang.duckdb.jdbc.url", jdbcUrl);
+ config.setProperty("wayang.duckdb.jdbc.user", "");
+ config.setProperty("wayang.duckdb.jdbc.password", "");
+ config.getMappingProvider().addAllToWhitelist(
+ Collections.singleton(new JoinFlattenMapping()));
+ return new WayangContext(config)
+ .withPlugin(DuckDB.plugin());
+ }
+
+ private TableSink tableSink(String... columnNames) {
+ return new TableSink<>(new Properties(), "overwrite", SINK_TABLE, columnNames);
+ }
+
+ private static MapOperator, Record> joinFlattenOperator() {
+ MapOperator, Record> operator = new MapOperator<>(
+ new TransformationDescriptor<>(
+ new JoinFlattenFunction(),
+ DataUnitType.createBasicUnchecked(Tuple2.class),
+ DataUnitType.createBasic(Record.class)),
+ DataSetType.createDefaultUnchecked(Tuple2.class),
+ DataSetType.createDefault(Record.class));
+ operator.setName(JOIN_FLATTEN_NAME);
+ return operator;
+ }
+
+ private static Record flattenJoinResult(Object joinResult) {
+ if (joinResult instanceof Record) {
+ return (Record) joinResult;
+ }
+ Tuple2, ?> pair = (Tuple2, ?>) joinResult;
+ Record left = (Record) pair.field0;
+ Record right = (Record) pair.field1;
+ return new Record(
+ left.getField(0),
+ left.getField(1),
+ left.getField(2),
+ left.getField(3),
+ right.getField(0),
+ right.getField(1),
+ right.getField(2));
+ }
+
+ private long queryLong(String sql) {
+ try (Connection connection = jdbc();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ resultSet.next();
+ return resultSet.getLong(1);
+ } catch (Exception e) {
+ throw new RuntimeException("query failed: " + sql, e);
+ }
+ }
+
+ private double queryDouble(String sql) {
+ try (Connection connection = jdbc();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ resultSet.next();
+ return resultSet.getDouble(1);
+ } catch (Exception e) {
+ throw new RuntimeException("query failed: " + sql, e);
+ }
+ }
+
+ private String queryString(String sql) {
+ try (Connection connection = jdbc();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ resultSet.next();
+ return resultSet.getString(1);
+ } catch (Exception e) {
+ throw new RuntimeException("query failed: " + sql, e);
+ }
+ }
+
+ private static String captureStdout(Runnable runnable) {
+ PrintStream originalOut = System.out;
+ ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ try (PrintStream capture = new PrintStream(buffer, true, StandardCharsets.UTF_8)) {
+ System.setOut(capture);
+ runnable.run();
+ } finally {
+ System.setOut(originalOut);
+ }
+ return buffer.toString(StandardCharsets.UTF_8);
+ }
+
+ private void assertSingleDoubleResult(double expected) {
+ try (Connection connection = jdbc();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery("SELECT * FROM " + SINK_TABLE)) {
+ assertTrue(resultSet.next());
+ assertEquals(expected, resultSet.getDouble(1), 0.01);
+ assertFalse(resultSet.next());
+ } catch (Exception e) {
+ throw new RuntimeException("query failed: SELECT * FROM " + SINK_TABLE, e);
+ }
+ }
+
+ private Map readRegionSums() {
+ Map sums = new HashMap<>();
+ try (Connection connection = jdbc();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery("SELECT * FROM " + SINK_TABLE)) {
+ while (resultSet.next()) {
+ sums.put(resultSet.getString(1), resultSet.getDouble(2));
+ }
+ return sums;
+ } catch (Exception e) {
+ throw new RuntimeException("query failed: SELECT * FROM " + SINK_TABLE, e);
+ }
+ }
+
+ private static final class JoinFlattenFunction implements
+ FunctionDescriptor.SerializableFunction, Record> {
+
+ @Override
+ public Record apply(Tuple2 tuple) {
+ return flattenJoinResult(tuple);
+ }
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static final class JoinFlattenMapping implements Mapping {
+
+ @Override
+ public java.util.Collection getTransformations() {
+ OperatorPattern pattern = new OperatorPattern(
+ "joinFlatten",
+ new MapOperator(null, DataSetType.none(), DataSetType.createDefault(Record.class)),
+ false)
+ .withAdditionalTest(operator -> JOIN_FLATTEN_NAME.equals(((MapOperator) operator).getName()));
+
+ ReplacementSubplanFactory factory = new ReplacementSubplanFactory.OfSingleOperators(
+ (matchedOperator, epoch) -> createDuckDBProjection().at(epoch));
+
+ return Collections.singleton(new PlanTransformation(
+ SubplanPattern.createSingleton(pattern),
+ factory,
+ DuckDBPlatform.getInstance()));
+ }
+
+ private static DuckDBProjectionOperator createDuckDBProjection() {
+ ProjectionDescriptor, Record> descriptor = new ProjectionDescriptor<>(
+ new JoinFlattenFunction(),
+ Arrays.asList(JOIN_COLUMNS),
+ DataUnitType.createBasicUnchecked(Tuple2.class),
+ DataUnitType.createBasic(Record.class));
+ MapOperator, Record> projection = new MapOperator<>(
+ descriptor,
+ DataSetType.createDefaultUnchecked(Tuple2.class),
+ DataSetType.createDefault(Record.class));
+ projection.setName(JOIN_FLATTEN_NAME);
+ return new DuckDBProjectionOperator((MapOperator) (MapOperator) projection);
+ }
+ }
+
+ private static Connection jdbc() throws Exception {
+ return DriverManager.getConnection(jdbcUrl);
+ }
+
+ private static String normalizeDuckDbUrl(String url) {
+ String prefix = "jdbc:duckdb:";
+ if (!url.startsWith(prefix) || url.length() == prefix.length()) {
+ return url;
+ }
+
+ String databasePath = url.substring(prefix.length());
+ Path path = Path.of(databasePath);
+ if (path.isAbsolute()) {
+ return url;
+ }
+
+ Path current = Path.of("").toAbsolutePath();
+ while (current != null) {
+ Path candidate = current.resolve(path).normalize();
+ if (Files.exists(candidate)) {
+ return prefix + candidate;
+ }
+ current = current.getParent();
+ }
+
+ return prefix + Path.of("").toAbsolutePath().resolve(path).normalize();
+ }
+}
diff --git a/wayang-platforms/wayang-duckdb/src/test/java/org/apache/wayang/duckdb/DuckDBParquetSourceIT.java b/wayang-platforms/wayang-duckdb/src/test/java/org/apache/wayang/duckdb/DuckDBParquetSourceIT.java
new file mode 100644
index 000000000..0ff93d36a
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/test/java/org/apache/wayang/duckdb/DuckDBParquetSourceIT.java
@@ -0,0 +1,197 @@
+/*
+ * 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.duckdb;
+
+import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.operators.TableSink;
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.api.WayangContext;
+import org.apache.wayang.core.plan.wayangplan.WayangPlan;
+import org.apache.wayang.duckdb.operators.DuckDBParquetSource;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Integration tests for {@link DuckDBParquetSource}.
+ */
+class DuckDBParquetSourceIT {
+
+ private static final String SCHEMA = "wayang_parquet_it";
+ private static final String SINK_TABLE = SCHEMA + ".orders_parquet_copy";
+ private static final String MAPPED_SINK_TABLE = SCHEMA + ".orders_mapped_copy";
+ private static final String GCS_SINK_TABLE = SCHEMA + ".orders_gcs_copy";
+ private static final String SOURCE_VIEW = SCHEMA + ".orders_parquet";
+ private static final String[] COLUMNS = {"order_id", "region", "amount"};
+ private static final String DEFAULT_GCS_URI =
+ "gs://anaconda-public-data/nyc-taxi/nyc.parquet/part.0.parquet";
+
+ private static Path databaseFile;
+ private static Path parquetFile;
+ private static String jdbcUrl;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ databaseFile = Files.createTempFile("wayang-duckdb-parquet-", ".duckdb");
+ parquetFile = Files.createTempFile("wayang-duckdb-orders-", ".parquet");
+ Files.deleteIfExists(databaseFile);
+ Files.deleteIfExists(parquetFile);
+ jdbcUrl = "jdbc:duckdb:" + databaseFile.toAbsolutePath();
+
+ try (Connection connection = jdbc(); Statement statement = connection.createStatement()) {
+ statement.execute("CREATE SCHEMA IF NOT EXISTS " + SCHEMA);
+ statement.execute("DROP TABLE IF EXISTS " + SINK_TABLE);
+ statement.execute("DROP TABLE IF EXISTS " + MAPPED_SINK_TABLE);
+ statement.execute("DROP TABLE IF EXISTS " + GCS_SINK_TABLE);
+ statement.execute("DROP VIEW IF EXISTS " + SOURCE_VIEW);
+ statement.execute("COPY ("
+ + "SELECT * FROM (VALUES "
+ + "(CAST(1 AS BIGINT), 'AMER', CAST(10.0 AS DOUBLE)), "
+ + "(CAST(2 AS BIGINT), 'EMEA', CAST(20.0 AS DOUBLE)), "
+ + "(CAST(3 AS BIGINT), 'APAC', CAST(30.0 AS DOUBLE))"
+ + ") AS t(order_id, region, amount)"
+ + ") TO '" + parquetUri() + "' (FORMAT PARQUET)");
+ statement.execute("CREATE VIEW " + SOURCE_VIEW + " AS SELECT * FROM read_parquet('"
+ + parquetUri() + "')");
+ }
+ }
+
+ @AfterAll
+ static void tearDown() throws Exception {
+ if (databaseFile != null) {
+ Files.deleteIfExists(databaseFile);
+ Files.deleteIfExists(databaseFile.resolveSibling(databaseFile.getFileName() + ".wal"));
+ }
+ if (parquetFile != null) {
+ Files.deleteIfExists(parquetFile);
+ }
+ }
+
+ @Test
+ void readsLocalParquetFileViaAutoCreatedDuckDbView() throws Exception {
+ DuckDBParquetSource source = new DuckDBParquetSource(parquetUri(), null, COLUMNS);
+ TableSink sink = new TableSink<>(new Properties(), "overwrite", SINK_TABLE, COLUMNS);
+ source.connectTo(0, sink, 0);
+
+ wayangContext(true, "").execute(new WayangPlan(sink));
+
+ assertEquals(3, queryLong("SELECT count(*) FROM " + SINK_TABLE));
+ assertEquals(60.0, queryDouble("SELECT sum(amount) FROM " + SINK_TABLE), 0.01);
+ }
+
+ @Test
+ void mapsParquetUriToExistingDuckDbRelation() throws Exception {
+ DuckDBParquetSource source = new DuckDBParquetSource(parquetUri(), null, COLUMNS);
+ TableSink sink = new TableSink<>(new Properties(), "overwrite", MAPPED_SINK_TABLE, COLUMNS);
+ source.connectTo(0, sink, 0);
+
+ Configuration configuration = baseConfiguration();
+ configuration.setProperty("wayang.duckdb.parquetsource.mappings", parquetUri() + "=" + SOURCE_VIEW);
+ new WayangContext(configuration).withPlugin(DuckDB.plugin()).execute(new WayangPlan(sink));
+
+ assertEquals(3, queryLong("SELECT count(*) FROM " + MAPPED_SINK_TABLE));
+ assertEquals(60.0, queryDouble("SELECT sum(amount) FROM " + MAPPED_SINK_TABLE), 0.01);
+ }
+
+ @Test
+ void readsPublicGcsParquetFileViaDuckDbHttpfs() throws Exception {
+ String gcsUri = System.getProperty("duckdb.gcs.parquet.uri",
+ System.getenv().getOrDefault("DUCKDB_GCS_PARQUET_URI", DEFAULT_GCS_URI));
+ Assumptions.assumeTrue(isGcsParquetReachable(gcsUri), "DuckDB httpfs cannot reach " + gcsUri);
+
+ DuckDBParquetSource source = new DuckDBParquetSource(gcsUri, null);
+ TableSink sink = new TableSink<>(new Properties(), "overwrite", GCS_SINK_TABLE);
+ source.connectTo(0, sink, 0);
+
+ wayangContext(true, "INSTALL httpfs; LOAD httpfs").execute(new WayangPlan(sink));
+
+ assertEquals(
+ queryLong("SELECT count(*) FROM read_parquet('" + gcsUri + "')"),
+ queryLong("SELECT count(*) FROM " + GCS_SINK_TABLE));
+ }
+
+ private WayangContext wayangContext(boolean autoCreate, String prepareSql) {
+ Configuration configuration = baseConfiguration();
+ configuration.setProperty("wayang.duckdb.parquetsource.auto-create", Boolean.toString(autoCreate));
+ if (!prepareSql.isEmpty()) {
+ configuration.setProperty("wayang.duckdb.parquetsource.prepare-sql", prepareSql);
+ }
+ return new WayangContext(configuration).withPlugin(DuckDB.plugin());
+ }
+
+ private static Configuration baseConfiguration() {
+ Configuration configuration = new Configuration();
+ configuration.setProperty("wayang.duckdb.jdbc.url", jdbcUrl);
+ configuration.setProperty("wayang.duckdb.jdbc.user", "");
+ configuration.setProperty("wayang.duckdb.jdbc.password", "");
+ return configuration;
+ }
+
+ private static boolean isGcsParquetReachable(String gcsUri) {
+ try (Connection connection = jdbc(); Statement statement = connection.createStatement()) {
+ statement.execute("INSTALL httpfs");
+ statement.execute("LOAD httpfs");
+ return queryLong(statement, "SELECT count(*) FROM read_parquet('" + gcsUri + "')") > 0;
+ } catch (Exception e) {
+ System.err.println("[DuckDBParquetSourceIT] GCS Parquet unavailable: " + e.getMessage());
+ return false;
+ }
+ }
+
+ private static long queryLong(String sql) throws Exception {
+ try (Connection connection = jdbc(); Statement statement = connection.createStatement()) {
+ return queryLong(statement, sql);
+ }
+ }
+
+ private static long queryLong(Statement statement, String sql) throws Exception {
+ try (ResultSet resultSet = statement.executeQuery(sql)) {
+ resultSet.next();
+ return resultSet.getLong(1);
+ }
+ }
+
+ private static double queryDouble(String sql) throws Exception {
+ try (Connection connection = jdbc();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ resultSet.next();
+ return resultSet.getDouble(1);
+ }
+ }
+
+ private static Connection jdbc() throws Exception {
+ return DriverManager.getConnection(jdbcUrl);
+ }
+
+ private static String parquetUri() {
+ return parquetFile.toAbsolutePath().toString().replace('\\', '/');
+ }
+}
diff --git a/wayang-platforms/wayang-duckdb/src/test/resources/duckdb-ga-smoke.properties b/wayang-platforms/wayang-duckdb/src/test/resources/duckdb-ga-smoke.properties
new file mode 100644
index 000000000..a7a4c1e57
--- /dev/null
+++ b/wayang-platforms/wayang-duckdb/src/test/resources/duckdb-ga-smoke.properties
@@ -0,0 +1,28 @@
+#
+# 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.
+#
+
+wayang.profiler.ga.timelimit.ms = 1000
+wayang.profiler.ga.maxgenerations = 1
+wayang.profiler.ga.maxstablegenerations = 1
+wayang.profiler.ga.superoptimizations = 1
+wayang.profiler.ga.intermediateupdate = 1
+wayang.profiler.ga.min-exec-time = 1
+wayang.profiler.ga.max-cardinality-spread = 100
+wayang.profiler.ga.min-cardinality-confidence = 0
+wayang.profiler.ga.binning = 1.0
+wayang.profiler.ga.output-file = wayang-platforms/wayang-duckdb/target/cost-profiling/duckdb-ga-smoke/learned.properties
+wayang.profiler.ga.noise-filter.max = 0
diff --git a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java
index dd5792f52..7cfbad784 100644
--- a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java
+++ b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java
@@ -33,7 +33,7 @@
import org.apache.wayang.basic.operators.JoinOperator;
import org.apache.wayang.basic.operators.SpatialFilterOperator;
import org.apache.wayang.basic.operators.SpatialJoinOperator;
-import org.apache.wayang.basic.operators.TableSource;
+import org.apache.wayang.core.api.Configuration;
import org.apache.wayang.core.api.Job;
import org.apache.wayang.core.api.exception.WayangException;
import org.apache.wayang.core.optimizer.OptimizationContext;
@@ -57,47 +57,61 @@
import org.apache.wayang.jdbc.operators.JdbcProjectionOperator;
import org.apache.wayang.jdbc.operators.JdbcReduceByOperator;
import org.apache.wayang.jdbc.operators.JdbcSortOperator;
+import org.apache.wayang.jdbc.operators.JdbcSourceOperator;
import org.apache.wayang.jdbc.operators.JdbcTableSinkOperator;
-import org.apache.wayang.jdbc.operators.JdbcTableSource;
import org.apache.wayang.jdbc.platform.JdbcPlatformTemplate;
/**
* {@link Executor} implementation for the {@link JdbcPlatformTemplate}.
*/
public class JdbcExecutor extends ExecutorTemplate {
- public static StringBuilder createSqlString(final JdbcExecutor jdbcExecutor, final JdbcTableSource tableOp,
+ public static StringBuilder createSqlString(final JdbcExecutor jdbcExecutor, final JdbcSourceOperator sourceOp,
final Collection filterTasks, final JdbcProjectionOperator projectionTask, final JdbcGlobalReduceOperator globalReduceTask, final JdbcReduceByOperator reduceByTask, final JdbcSortOperator sortTask,
final Collection joinTasks) {
- final String tableName = tableOp.createSqlClause(jdbcExecutor.connection, jdbcExecutor.functionCompiler);
+ return createSqlString(jdbcExecutor, sourceOp, filterTasks, projectionTask, globalReduceTask, reduceByTask,
+ sortTask, joinTasks, null);
+ }
+
+ public static StringBuilder createSqlString(final JdbcExecutor jdbcExecutor, final JdbcSourceOperator sourceOp,
+ final Collection filterTasks, final JdbcProjectionOperator projectionTask, final JdbcGlobalReduceOperator globalReduceTask, final JdbcReduceByOperator reduceByTask, final JdbcSortOperator sortTask,
+ final Collection joinTasks, final Configuration configuration) {
+ final String sourceName = sourceOp.createSqlClause(
+ jdbcExecutor.connection,
+ jdbcExecutor.functionCompiler,
+ configuration
+ );
final Collection conditions = filterTasks.stream()
- .map(op -> op.createSqlClause(jdbcExecutor.connection, jdbcExecutor.functionCompiler))
+ .map(op -> op.createSqlClause(jdbcExecutor.connection, jdbcExecutor.functionCompiler, configuration))
.collect(Collectors.toList());
final Collection joins = joinTasks.stream()
- .map(op -> op.createSqlClause(jdbcExecutor.connection, jdbcExecutor.functionCompiler))
+ .map(op -> op.createSqlClause(jdbcExecutor.connection, jdbcExecutor.functionCompiler, configuration))
.collect(Collectors.toList());
final String selectClause;
if (globalReduceTask != null) {
selectClause = globalReduceTask.createSqlClause(
jdbcExecutor.connection,
- jdbcExecutor.functionCompiler
+ jdbcExecutor.functionCompiler,
+ configuration
);
} else if (reduceByTask != null) {
selectClause = reduceByTask.createSqlClause(
jdbcExecutor.connection,
- jdbcExecutor.functionCompiler
+ jdbcExecutor.functionCompiler,
+ configuration
);
} else if (projectionTask != null) {
selectClause = projectionTask.createSqlClause(
jdbcExecutor.connection,
- jdbcExecutor.functionCompiler
+ jdbcExecutor.functionCompiler,
+ configuration
);
} else {
selectClause = "*";
}
final StringBuilder sb = new StringBuilder(1000);
- sb.append("SELECT ").append(selectClause).append(" FROM ").append(tableName);
+ sb.append("SELECT ").append(selectClause).append(" FROM ").append(sourceName);
if (!joins.isEmpty()) {
final String separator = " ";
for (final String join : joins) {
@@ -114,7 +128,8 @@ public static StringBuilder createSqlString(final JdbcExecutor jdbcExecutor, fin
if (sortTask != null) {
sb.append(sortTask.createSqlClause(
jdbcExecutor.connection,
- jdbcExecutor.functionCompiler
+ jdbcExecutor.functionCompiler,
+ configuration
));
}
@@ -136,12 +151,13 @@ protected static Tuple2 createSqlQuery(final E
final Collection> startTasks = stage.getStartTasks();
// Verify that we can handle this instance.
- final ExecutionTask startTask = JdbcExecutor.selectStartTask(startTasks, stage);
- assert startTask.getOperator() instanceof TableSource
- : "Invalid JDBC stage: Start task has to be a TableSource";
+ JdbcExecutor.prepareSourceTasks(startTasks, jdbcExecutor, context.getConfiguration());
+ final ExecutionTask startTask = JdbcExecutor.selectStartTask(startTasks, stage, context.getConfiguration());
+ assert startTask.getOperator() instanceof JdbcSourceOperator
+ : "Invalid JDBC stage: Start task has to be a JDBC source";
// Extract the different types of ExecutionOperators from the stage.
- final JdbcTableSource tableOp = (JdbcTableSource) startTask.getOperator();
+ final JdbcSourceOperator sourceOp = (JdbcSourceOperator) startTask.getOperator();
SqlQueryChannel.Instance tipChannelInstance = JdbcExecutor.instantiateOutboundChannel(startTask, context,
jdbcExecutor);
final Collection filterTasks = new ArrayList<>(4);
@@ -185,7 +201,8 @@ protected static Tuple2 createSqlQuery(final E
}
// Create the SQL query.
- final StringBuilder query = createSqlString(jdbcExecutor, tableOp, filterTasks, projectionTask, globalReduceTask, reduceByTask, sortTask, joinTasks);
+ final StringBuilder query = createSqlString(jdbcExecutor, sourceOp, filterTasks, projectionTask,
+ globalReduceTask, reduceByTask, sortTask, joinTasks, context.getConfiguration());
return new Tuple2<>(query.toString(), tipChannelInstance);
}
@@ -196,6 +213,12 @@ protected static Tuple2 createSqlQuery(final E
* assumes its first key descriptor's table is used in the {@code FROM} clause.
*/
private static ExecutionTask selectStartTask(final Collection> startTasks, final ExecutionStage stage) {
+ return selectStartTask(startTasks, stage, null);
+ }
+
+ private static ExecutionTask selectStartTask(final Collection> startTasks,
+ final ExecutionStage stage,
+ final Configuration configuration) {
if (startTasks.size() == 1) {
return (ExecutionTask) startTasks.iterator().next();
}
@@ -206,9 +229,12 @@ private static ExecutionTask selectStartTask(final Collection> startTasks, fin
final String leftTableName = joinOperator.getKeyDescriptor0().getSqlImplementation().field0;
for (Object startTaskObject : startTasks) {
final ExecutionTask startTask = (ExecutionTask) startTaskObject;
- if (startTask.getOperator() instanceof JdbcTableSource
- && ((JdbcTableSource) startTask.getOperator()).getTableName().equals(leftTableName)) {
- return startTask;
+ if (startTask.getOperator() instanceof JdbcSourceOperator) {
+ final JdbcSourceOperator sourceOperator = (JdbcSourceOperator) startTask.getOperator();
+ if (sourceOperator.getSourceName().equals(leftTableName)
+ || sourceOperator.getSourceName(configuration).equals(leftTableName)) {
+ return startTask;
+ }
}
}
}
@@ -217,6 +243,21 @@ private static ExecutionTask selectStartTask(final Collection> startTasks, fin
throw new WayangException("Could not determine the left table source for JDBC stage.");
}
+ private static void prepareSourceTasks(final Collection> startTasks,
+ final JdbcExecutor jdbcExecutor,
+ final Configuration configuration) {
+ for (Object startTaskObject : startTasks) {
+ final ExecutionTask startTask = (ExecutionTask) startTaskObject;
+ if (startTask.getOperator() instanceof JdbcSourceOperator) {
+ ((JdbcSourceOperator) startTask.getOperator()).prepareSource(
+ jdbcExecutor.connection,
+ jdbcExecutor.functionCompiler,
+ configuration
+ );
+ }
+ }
+ }
+
/**
* Handles execution stages that end with a {@link JdbcTableSinkOperator}.
* Composes a SQL query from the stage's operators and executes it directly on
@@ -230,17 +271,22 @@ private static long executeSinkStage(final ExecutionStage stage, final Optimizat
final JdbcExecutor jdbcExecutor) {
final Collection> startTasks = stage.getStartTasks();
final Collection> termTasks = stage.getTerminalTasks();
+ JdbcExecutor.prepareSourceTasks(startTasks, jdbcExecutor, optimizationContext.getConfiguration());
- final ExecutionTask startTask = JdbcExecutor.selectStartTask(startTasks, stage);
+ final ExecutionTask startTask = JdbcExecutor.selectStartTask(
+ startTasks,
+ stage,
+ optimizationContext.getConfiguration()
+ );
assert termTasks.size() == 1 : "Invalid JDBC stage: multiple terminal tasks are not currently supported.";
final ExecutionTask termTask = (ExecutionTask) termTasks.toArray()[0];
- assert startTask.getOperator() instanceof TableSource
- : "Invalid JDBC stage: Start task has to be a TableSource";
+ assert startTask.getOperator() instanceof JdbcSourceOperator
+ : "Invalid JDBC stage: Start task has to be a JDBC source";
assert termTask.getOperator() instanceof JdbcTableSinkOperator
: "Invalid JDBC stage: Terminal task has to be a JdbcTableSinkOperator";
// Extract operators from the stage
- final JdbcTableSource tableOp = (JdbcTableSource) startTask.getOperator();
+ final JdbcSourceOperator sourceOp = (JdbcSourceOperator) startTask.getOperator();
final JdbcTableSinkOperator sinkOp = (JdbcTableSinkOperator) termTask.getOperator();
final Collection filterTasks = new ArrayList<>(4);
JdbcProjectionOperator projectionTask = null;
@@ -275,14 +321,8 @@ private static long executeSinkStage(final ExecutionStage stage, final Optimizat
}
// Compose the SELECT query
- final StringBuilder selectQuery = createSqlString(jdbcExecutor, tableOp, filterTasks, projectionTask,
- globalReduceTask, reduceByTask, sortTask, joinTasks);
-
- // Remove trailing semicolon from SELECT
- String selectSql = selectQuery.toString();
- if (selectSql.endsWith(";")) {
- selectSql = selectSql.substring(0, selectSql.length() - 1);
- }
+ final String selectSql = createSqlString(jdbcExecutor, sourceOp, filterTasks, projectionTask,
+ globalReduceTask, reduceByTask, sortTask, joinTasks, optimizationContext.getConfiguration()).toString();
// Get the sink's SQL clause
final String sinkClause = sinkOp.createSqlClause(jdbcExecutor.connection, jdbcExecutor.functionCompiler);
@@ -348,7 +388,7 @@ private static ExecutionTask findJdbcExecutionOperatorTaskInStage(final Executio
assert task.getNumOuputChannels() == 1;
final Channel outputChannel = task.getOutputChannel(0);
final ExecutionTask consumer = WayangCollections.getSingle(outputChannel.getConsumers());
- return consumer.getStage() == stage && consumer.getOperator() instanceof JdbcExecutionOperator
+ return consumer.getStage() == stage && consumer.getOperator() instanceof JdbcExecutionOperator
? consumer
: null;
}
diff --git a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcExecutionOperator.java b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcExecutionOperator.java
index 570897aed..e8457f508 100644
--- a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcExecutionOperator.java
+++ b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcExecutionOperator.java
@@ -19,6 +19,7 @@
package org.apache.wayang.jdbc.operators;
import org.apache.wayang.basic.operators.TableSource;
+import org.apache.wayang.core.api.Configuration;
import org.apache.wayang.core.plan.wayangplan.ExecutionOperator;
import org.apache.wayang.core.platform.ChannelDescriptor;
import org.apache.wayang.jdbc.compiler.FunctionCompiler;
@@ -42,6 +43,13 @@ public interface JdbcExecutionOperator extends ExecutionOperator {
*/
String createSqlClause(Connection connection, FunctionCompiler compiler);
+ /**
+ * Creates a SQL clause under the given configuration.
+ */
+ default String createSqlClause(Connection connection, FunctionCompiler compiler, Configuration configuration) {
+ return this.createSqlClause(connection, compiler);
+ }
+
@Override
JdbcPlatformTemplate getPlatform();
diff --git a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcJoinOperator.java b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcJoinOperator.java
index 6ef378422..1368dde66 100644
--- a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcJoinOperator.java
+++ b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcJoinOperator.java
@@ -66,11 +66,17 @@ public JdbcJoinOperator(JoinOperator that) {
@Override
public String createSqlClause(Connection connection, FunctionCompiler compiler) {
+ return this.createSqlClause(connection, compiler, null);
+ }
+
+ @Override
+ public String createSqlClause(Connection connection, FunctionCompiler compiler, Configuration configuration) {
final Tuple left = this.keyDescriptor0.getSqlImplementation();
final Tuple right = this.keyDescriptor1.getSqlImplementation();
- final String leftTableName = left.field0;
+ final String platformId = this.getPlatform().getPlatformId();
+ final String leftTableName = JdbcParquetSource.resolveSourceName(configuration, platformId, left.field0);
final String leftKeys = left.field1;
- final String rightTableName = right.field0;
+ final String rightTableName = JdbcParquetSource.resolveSourceName(configuration, platformId, right.field0);
final String rightKeys = right.field1;
if (leftKeys.contains(",") && rightKeys.contains(",")) {
diff --git a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcParquetSource.java b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcParquetSource.java
new file mode 100644
index 000000000..4bea068df
--- /dev/null
+++ b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcParquetSource.java
@@ -0,0 +1,410 @@
+/*
+ * 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.jdbc.operators;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+import org.apache.wayang.basic.operators.ParquetSource;
+import org.apache.wayang.commons.util.profiledb.model.measurement.TimeMeasurement;
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.api.exception.WayangException;
+import org.apache.wayang.core.optimizer.OptimizationContext;
+import org.apache.wayang.core.optimizer.cardinality.CardinalityEstimate;
+import org.apache.wayang.jdbc.compiler.FunctionCompiler;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * JDBC implementation for {@link ParquetSource}s exposed by an engine as a SQL
+ * relation, such as a Hive/Iceberg table, BigQuery external table, or DuckDB
+ * view over {@code read_parquet(...)}.
+ */
+public abstract class JdbcParquetSource extends ParquetSource implements JdbcSourceOperator {
+
+ private static final Pattern PROPERTY_PLACEHOLDER = Pattern.compile("\\$\\{(env|sys):([^}]+)}");
+
+ public JdbcParquetSource(String sourceName, String[] projection, String... columnNames) {
+ super(sourceName, projection, columnNames);
+ }
+
+ public JdbcParquetSource(ParquetSource that) {
+ super(that);
+ }
+
+ @Override
+ public String getSourceName() {
+ return this.getInputUrl();
+ }
+
+ @Override
+ public String getSourceName(Configuration configuration) {
+ return this.resolveSourceName(configuration);
+ }
+
+ @Override
+ public String createSqlClause(Connection connection, FunctionCompiler compiler) {
+ return this.getSourceName();
+ }
+
+ @Override
+ public String createSqlClause(Connection connection, FunctionCompiler compiler, Configuration configuration) {
+ return this.resolveSourceName(connection, configuration);
+ }
+
+ @Override
+ public void prepareSource(Connection connection, FunctionCompiler compiler, Configuration configuration) {
+ this.resolveSourceName(connection, configuration);
+ }
+
+ @Override
+ public String getLoadProfileEstimatorConfigurationKey() {
+ return String.format("wayang.%s.parquetsource.load", this.getPlatform().getPlatformId());
+ }
+
+ @Override
+ public org.apache.wayang.core.optimizer.cardinality.CardinalityEstimator getCardinalityEstimator(int outputIndex) {
+ assert outputIndex == 0;
+ return new org.apache.wayang.core.optimizer.cardinality.CardinalityEstimator() {
+ @Override
+ public CardinalityEstimate estimate(OptimizationContext optimizationContext,
+ CardinalityEstimate... inputEstimates) {
+ final TimeMeasurement timeMeasurement = optimizationContext.getJob().getStopWatch().start(
+ "Optimization", "Cardinality&Load Estimation", "Push Estimation", "Estimate source cardinalities"
+ );
+
+ try (Connection connection = JdbcParquetSource.this.getPlatform()
+ .createDatabaseDescriptor(optimizationContext.getConfiguration())
+ .createJdbcConnection()) {
+ final String sql = String.format("SELECT count(*) FROM %s",
+ JdbcParquetSource.this.resolveSourceName(
+ connection,
+ optimizationContext.getConfiguration()
+ ));
+ final ResultSet resultSet = connection.createStatement().executeQuery(sql);
+ if (!resultSet.next()) {
+ throw new SQLException("No query result for \"" + sql + "\".");
+ }
+ long cardinality = resultSet.getLong(1);
+ return new CardinalityEstimate(cardinality, cardinality, 1d);
+ } catch (Exception e) {
+ LogManager.getLogger(this.getClass()).error(
+ "Could not estimate cardinality for {}.", JdbcParquetSource.this, e
+ );
+ return new CardinalityEstimate(10, 10000000, 0.9);
+ } finally {
+ timeMeasurement.stop();
+ }
+ }
+ };
+ }
+
+ @Override
+ public Optional createCardinalityEstimator(
+ int outputIndex,
+ Configuration configuration) {
+ return Optional.of(this.getCardinalityEstimator(outputIndex));
+ }
+
+ private String resolveSourceName(Configuration configuration) {
+ if (configuration == null) {
+ return this.getSourceName();
+ }
+
+ final String platformId = this.getPlatform().getPlatformId();
+ return resolveSourceName(configuration, platformId, this.getInputUrl());
+ }
+
+ private String resolveSourceName(Connection connection, Configuration configuration) {
+ final String sourceName = this.resolveSourceName(configuration);
+ if (configuration == null) {
+ return sourceName;
+ }
+
+ final String platformId = this.getPlatform().getPlatformId();
+ this.executePrepareSql(connection, configuration, platformId);
+ if (isAutoCreateEnabled(configuration, platformId)) {
+ this.createExternalRelation(connection, configuration, platformId, sourceName);
+ }
+
+ return sourceName;
+ }
+
+ public static String resolveSourceName(Configuration configuration, String platformId, String inputUrl) {
+ if (configuration == null) {
+ return inputUrl;
+ }
+
+ return findMappedRelation(configuration, platformId, inputUrl)
+ .orElseGet(() -> isAutoCreateEnabled(configuration, platformId) && isParquetLocation(inputUrl)
+ ? createGeneratedRelationName(configuration, platformId, inputUrl)
+ : inputUrl);
+ }
+
+ private static boolean isParquetLocation(String inputUrl) {
+ return inputUrl.contains("://")
+ || inputUrl.startsWith("/")
+ || inputUrl.startsWith("\\")
+ || inputUrl.matches("^[A-Za-z]:[\\\\/].*")
+ || inputUrl.contains(".parquet");
+ }
+
+ private static Optional findMappedRelation(Configuration configuration, String platformId, String inputUrl) {
+ final String mappingKey = String.format("wayang.%s.parquetsource.mappings", platformId);
+ final Optional mapping = configuration.getOptionalStringProperty(mappingKey);
+ if (mapping.isEmpty()) {
+ return Optional.empty();
+ }
+
+ for (String entry : mapping.get().split(";")) {
+ final String trimmedEntry = entry.trim();
+ if (trimmedEntry.isEmpty()) {
+ continue;
+ }
+
+ final int separator = trimmedEntry.indexOf('=');
+ if (separator < 0) {
+ LogManager.getLogger(JdbcParquetSource.class).warn(
+ "Ignoring invalid Parquet source mapping entry '{}' for {}.", trimmedEntry, mappingKey
+ );
+ continue;
+ }
+
+ final String sourceUri = trimmedEntry.substring(0, separator).trim();
+ final String relationName = trimmedEntry.substring(separator + 1).trim();
+ if (sourceUri.equals(inputUrl) && !relationName.isEmpty()) {
+ return Optional.of(relationName);
+ }
+ }
+
+ return Optional.empty();
+ }
+
+ private void executePrepareSql(Connection connection, Configuration configuration, String platformId) {
+ final String prepareSqlKey = String.format("wayang.%s.parquetsource.prepare-sql", platformId);
+ final Optional optionalPrepareSql = configuration.getOptionalStringProperty(prepareSqlKey);
+ if (optionalPrepareSql.isEmpty() || optionalPrepareSql.get().trim().isEmpty()) {
+ return;
+ }
+
+ try (Statement statement = connection.createStatement()) {
+ for (String sql : splitSqlStatements(resolvePlaceholders(optionalPrepareSql.get()))) {
+ statement.execute(sql);
+ }
+ } catch (SQLException e) {
+ throw new WayangException(String.format(
+ "Could not execute Parquet source prepare SQL configured by '%s'.",
+ prepareSqlKey
+ ), e);
+ }
+ }
+
+ private static List splitSqlStatements(String sql) {
+ final List statements = new ArrayList<>();
+ final StringBuilder current = new StringBuilder(sql.length());
+ boolean insideSingleQuote = false;
+ for (int i = 0; i < sql.length(); i++) {
+ char currentChar = sql.charAt(i);
+ if (currentChar == '\'') {
+ current.append(currentChar);
+ if (insideSingleQuote && i + 1 < sql.length() && sql.charAt(i + 1) == '\'') {
+ current.append(sql.charAt(++i));
+ } else {
+ insideSingleQuote = !insideSingleQuote;
+ }
+ } else if (currentChar == ';' && !insideSingleQuote) {
+ addStatement(statements, current);
+ } else {
+ current.append(currentChar);
+ }
+ }
+ addStatement(statements, current);
+ return statements;
+ }
+
+ private static void addStatement(List statements, StringBuilder statement) {
+ final String sql = statement.toString().trim();
+ if (!sql.isEmpty()) {
+ statements.add(sql);
+ }
+ statement.setLength(0);
+ }
+
+ private static String resolvePlaceholders(String sql) {
+ Matcher matcher = PROPERTY_PLACEHOLDER.matcher(sql);
+ StringBuffer resolved = new StringBuffer();
+ while (matcher.find()) {
+ final String type = matcher.group(1);
+ final String name = matcher.group(2);
+ final String value = "env".equals(type) ? System.getenv(name) : System.getProperty(name);
+ if (value == null) {
+ throw new WayangException(String.format(
+ "Could not resolve Parquet source prepare SQL placeholder '${%s:%s}'.",
+ type,
+ name
+ ));
+ }
+ matcher.appendReplacement(resolved, Matcher.quoteReplacement(value));
+ }
+ matcher.appendTail(resolved);
+ return resolved.toString();
+ }
+
+ private static boolean isAutoCreateEnabled(Configuration configuration, String platformId) {
+ return configuration.getBooleanProperty(
+ String.format("wayang.%s.parquetsource.auto-create", platformId),
+ false
+ );
+ }
+
+ private static String createGeneratedRelationName(Configuration configuration, String platformId, String inputUrl) {
+ final String prefix = configuration.getStringProperty(
+ String.format("wayang.%s.parquetsource.auto-create.relation-prefix", platformId),
+ "wayang_parquet_"
+ );
+ return prefix + shortHash(inputUrl);
+ }
+
+ private static String shortHash(String value) {
+ try {
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ byte[] hash = digest.digest(value.getBytes(StandardCharsets.UTF_8));
+ StringBuilder sb = new StringBuilder(16);
+ for (int i = 0; i < 8; i++) {
+ sb.append(String.format("%02x", hash[i]));
+ }
+ return sb.toString();
+ } catch (NoSuchAlgorithmException e) {
+ throw new WayangException("Could not create stable Parquet relation name.", e);
+ }
+ }
+
+ private void createExternalRelation(Connection connection,
+ Configuration configuration,
+ String platformId,
+ String relationName) {
+ final String templateKey = String.format("wayang.%s.parquetsource.auto-create.template", platformId);
+ final Optional optionalTemplate = configuration.getOptionalStringProperty(templateKey);
+ if (optionalTemplate.isEmpty()) {
+ throw new WayangException(String.format(
+ "Parquet auto-create is enabled for platform '%s', but '%s' is not configured.",
+ platformId,
+ templateKey
+ ));
+ }
+
+ final String template = optionalTemplate.get();
+ final String ddl = template
+ .replace("${relation}", relationName)
+ .replace("${uri}", this.escapeSqlString(this.getInputUrl()))
+ .replace("${columns}", this.createColumnDefinitions(templateKey, template));
+
+ try (Statement statement = connection.createStatement()) {
+ statement.execute(ddl);
+ } catch (SQLException e) {
+ throw new WayangException(String.format(
+ "Could not create Parquet SQL relation '%s' for '%s'.",
+ relationName,
+ this.getInputUrl()
+ ), e);
+ }
+ }
+
+ private String escapeSqlString(String value) {
+ return value.replace("'", "''");
+ }
+
+ private String createColumnDefinitions(String templateKey, String template) {
+ if (!template.contains("${columns}")) {
+ return "";
+ }
+
+ if (this.getSchema() == null || this.getSchema().getFields().isEmpty()) {
+ throw new WayangException(String.format(
+ "Parquet source auto-create template '%s' uses ${columns}, but no Parquet schema is available. "
+ + "Create the source with ParquetSource.create(...) or configure a template that does not "
+ + "need explicit columns.",
+ templateKey
+ ));
+ }
+
+ return this.getSchema().getFields().stream()
+ .map(field -> field.getName() + " " + this.toSqlType(field))
+ .reduce((left, right) -> left + ", " + right)
+ .orElse("");
+ }
+
+ private String toSqlType(Type field) {
+ if (!field.isPrimitive()) {
+ return "VARCHAR";
+ }
+
+ final PrimitiveType primitiveType = field.asPrimitiveType();
+ final LogicalTypeAnnotation logicalType = primitiveType.getLogicalTypeAnnotation();
+ if (logicalType instanceof LogicalTypeAnnotation.StringLogicalTypeAnnotation
+ || logicalType instanceof LogicalTypeAnnotation.EnumLogicalTypeAnnotation
+ || logicalType instanceof LogicalTypeAnnotation.UUIDLogicalTypeAnnotation) {
+ return "VARCHAR";
+ }
+ if (logicalType instanceof LogicalTypeAnnotation.DecimalLogicalTypeAnnotation) {
+ LogicalTypeAnnotation.DecimalLogicalTypeAnnotation decimal =
+ (LogicalTypeAnnotation.DecimalLogicalTypeAnnotation) logicalType;
+ return String.format("DECIMAL(%d,%d)", decimal.getPrecision(), decimal.getScale());
+ }
+ if (logicalType instanceof LogicalTypeAnnotation.DateLogicalTypeAnnotation) {
+ return "DATE";
+ }
+ if (logicalType instanceof LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) {
+ return "TIMESTAMP";
+ }
+
+ switch (primitiveType.getPrimitiveTypeName()) {
+ case BOOLEAN:
+ return "BOOLEAN";
+ case INT32:
+ return "INTEGER";
+ case INT64:
+ return "BIGINT";
+ case FLOAT:
+ return "REAL";
+ case DOUBLE:
+ return "DOUBLE";
+ case BINARY:
+ return "VARCHAR";
+ case FIXED_LEN_BYTE_ARRAY:
+ return "VARBINARY";
+ case INT96:
+ return "TIMESTAMP";
+ default:
+ return "VARCHAR";
+ }
+ }
+}
diff --git a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcSourceOperator.java b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcSourceOperator.java
new file mode 100644
index 000000000..b7934ac89
--- /dev/null
+++ b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcSourceOperator.java
@@ -0,0 +1,66 @@
+/*
+ * 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.jdbc.operators;
+
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.platform.ChannelDescriptor;
+import org.apache.wayang.jdbc.compiler.FunctionCompiler;
+
+import java.sql.Connection;
+import java.util.List;
+
+/**
+ * Marks JDBC operators that can start a SQL stage and provide a relation for a
+ * {@code FROM} clause.
+ */
+public interface JdbcSourceOperator extends JdbcExecutionOperator {
+
+ /**
+ * Name or expression used to identify this source in SQL metadata, e.g., in
+ * join descriptors.
+ */
+ String getSourceName();
+
+ /**
+ * Name or expression used for this source under the given configuration.
+ */
+ default String getSourceName(Configuration configuration) {
+ return this.getSourceName();
+ }
+
+ /**
+ * Creates a SQL clause for this source under the given configuration.
+ */
+ default String createSqlClause(Connection connection, FunctionCompiler compiler, Configuration configuration) {
+ return this.createSqlClause(connection, compiler);
+ }
+
+ /**
+ * Prepares this source for SQL generation, e.g., by registering a temporary
+ * relation. Implementations can keep this as a no-op when no preparation is
+ * required.
+ */
+ default void prepareSource(Connection connection, FunctionCompiler compiler, Configuration configuration) {
+ }
+
+ @Override
+ default List getSupportedInputChannels(int index) {
+ throw new UnsupportedOperationException("JDBC source operators have no input channels.");
+ }
+}
diff --git a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcTableSource.java b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcTableSource.java
index 4d7096649..ac414b564 100644
--- a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcTableSource.java
+++ b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/JdbcTableSource.java
@@ -33,7 +33,7 @@
/**
* PostgreSQL implementation for the {@link TableSource}.
*/
-public abstract class JdbcTableSource extends TableSource implements JdbcExecutionOperator {
+public abstract class JdbcTableSource extends TableSource implements JdbcSourceOperator {
/**
* Creates a new instance.
@@ -58,6 +58,10 @@ public String createSqlClause(Connection connection, FunctionCompiler compiler)
return this.getTableName();
}
+ @Override
+ public String getSourceName() {
+ return this.getTableName();
+ }
@Override
public String getLoadProfileEstimatorConfigurationKey() {
diff --git a/wayang-profiler/duckdb.md b/wayang-profiler/duckdb.md
new file mode 100644
index 000000000..a0fb9ecbf
--- /dev/null
+++ b/wayang-profiler/duckdb.md
@@ -0,0 +1,46 @@
+
+
+# DuckDB Cost Calibration
+
+Generate execution logs using `DuckDBCostPilotIT` as described in the
+[platform README](../wayang-platforms/wayang-duckdb/README.md#cost-profiling).
+The pilot remains a platform test; the genetic optimizer is provided by this
+module.
+
+From the repository root, install the profiler with the optional DuckDB runtime
+(the profile makes DuckDB operators available when reading execution logs):
+
+```sh
+./mvnw -Pskip-prerequisite-check,duckdb -pl wayang-profiler -am -DskipTests -Dpython.worker.tests.skip=true install
+```
+
+Run the existing optimizer with the calibration configuration and execution log:
+
+```sh
+./mvnw -Pskip-prerequisite-check,duckdb -pl wayang-profiler exec:java -Dexec.mainClass=org.apache.wayang.profiler.log.GeneticOptimizerApp "-Dexec.args=file:///absolute/path/to/wayang/wayang-profiler/src/main/resources/duckdb-ga.properties wayang-platforms/wayang-duckdb/target/cost-profiling/duckdb/executions.json"
+```
+
+On Windows, replace `./mvnw` with `.\mvnw.cmd`. Paths are relative to the
+repository root. Replace the configuration URL with the absolute file URL for
+your checkout (on Windows, for example, `file:///C:/src/wayang/wayang-profiler/src/main/resources/duckdb-ga.properties`). No platform-specific launch script or manually assembled Java
+classpath is required.
+
+The settings in `src/main/resources/duckdb-ga.properties` control the run limits
+and output location. By default, learned coefficients are written to
+`wayang-platforms/wayang-duckdb/target/cost-profiling/duckdb/learned-duckdb-relaxed.properties`.
+The optimizer is stochastic, so successive runs can produce different coefficients.
diff --git a/wayang-profiler/pom.xml b/wayang-profiler/pom.xml
index 77fa88ec5..024740e5a 100644
--- a/wayang-profiler/pom.xml
+++ b/wayang-profiler/pom.xml
@@ -100,4 +100,31 @@