Expand Beam SQL TPC-DS query support and add Python TPC-DS suite

.github/trigger_files/beam_PostCommit_Java_Tpcds_Dataflow.json

@@ -0,0 +1 @@
+{}

.github/trigger_files/beam_PostCommit_Java_Tpcds_Flink.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 1
+  "modification": 2
 }

.github/trigger_files/beam_PostCommit_Java_Tpcds_Spark.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 1
+  "modification": 2
 }

.github/workflows/README.md

@@ -372,6 +372,7 @@ PostCommit Jobs run in a schedule against master branch and generally do not get
 | [ PostCommit Python Examples Spark ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Examples_Spark.yml) | ['3.9','3.11','3.12'] |`beam_PostCommit_Python_Examples_Spark.json`| [![.github/workflows/beam_PostCommit_Python_Examples_Spark.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Examples_Spark.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Examples_Spark.yml?query=event%3Aschedule) |
 | [ PostCommit Python MongoDBIO IT ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_MongoDBIO_IT.yml) | N/A |`beam_PostCommit_Python_MongoDBIO_IT.json`| [![.github/workflows/beam_PostCommit_Python_MongoDBIO_IT.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_MongoDBIO_IT.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_MongoDBIO_IT.yml?query=event%3Aschedule) |
 | [ PostCommit Python Nexmark Direct ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Nexmark_Direct.yml) | N/A |`beam_PostCommit_Python_Nexmark_Direct.json`| [![.github/workflows/beam_PostCommit_Python_Nexmark_Direct.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Nexmark_Direct.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Nexmark_Direct.yml?query=event%3Aschedule) |
+| [ PostCommit Python Tpcds Prism ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Tpcds_Prism.yml) | N/A |`beam_PostCommit_Python_Tpcds_Prism.json`| [![.github/workflows/beam_PostCommit_Python_Tpcds_Prism.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Tpcds_Prism.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_Tpcds_Prism.yml?query=event%3Aschedule) |
 | [ PostCommit Python ValidatesContainer Dataflow ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_ValidatesContainer_Dataflow.yml) | ['3.9','3.10','3.11','3.12'] |`beam_PostCommit_Python_ValidatesContainer_Dataflow.json`| [![.github/workflows/beam_PostCommit_Python_ValidatesContainer_Dataflow.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_ValidatesContainer_Dataflow.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_ValidatesContainer_Dataflow.yml?query=event%3Aschedule) |
 | [ PostCommit Python ValidatesContainer Dataflow With RC ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_ValidatesContainer_Dataflow_With_RC.yml) | ['3.9','3.10','3.11','3.12'] |`beam_PostCommit_Python_ValidatesContainer_Dataflow_With_RC.json`| [![.github/workflows/beam_PostCommit_Python_ValidatesContainer_Dataflow_With_RC.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_ValidatesContainer_Dataflow_With_RC.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_ValidatesContainer_Dataflow_With_RC.yml?query=event%3Aschedule) |
 | [ PostCommit Python ValidatesRunner Dataflow ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_ValidatesRunner_Dataflow.yml) | ['3.9','3.11','3.12'] |`beam_PostCommit_Python_ValidatesRunner_Dataflow.json`| [![.github/workflows/beam_PostCommit_Python_ValidatesRunner_Dataflow.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_ValidatesRunner_Dataflow.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Python_ValidatesRunner_Dataflow.yml?query=event%3Aschedule) |

.github/workflows/beam_PostCommit_Java_Tpcds_Dataflow.yml

@@ -17,7 +17,7 @@ name: PostCommit Java Tpcds Dataflow
 
 on:
   schedule:
-    - cron: '30 4/6 * * *'
+    - cron: '30 4/8 * * *'
   pull_request:
     paths: ['release/trigger_all_tests.json', '.github/trigger_files/beam_PostCommit_Java_Tpcds_Dataflow.json']
   workflow_dispatch:
@@ -54,8 +54,8 @@ env:
     --dataSize=1GB 
     --sourceType=PARQUET 
     --dataDirectory=gs://beam-tpcds/datasets/parquet/nonpartitioned 
-    --resultsDirectory=gs://beam-tpcds/results/dataflow/ 
-    --tpcParallel=1
+    --resultsDirectory=gs://temp-storage-for-perf-tests/tpcds/results/dataflow
+    --tpcParallel=4
   INFLUXDB_USER: ${{ secrets.INFLUXDB_USER }}
   INFLUXDB_USER_PASSWORD: ${{ secrets.INFLUXDB_USER_PASSWORD }}
   tpcdsBigQueryArgs: |
@@ -71,7 +71,7 @@ env:
     --baseInfluxMeasurement=tpcds 
     --exportSummaryToInfluxDB=true 
     --influxRetentionPolicy=forever 
-  tpcdsQueriesArg: 3,7,10,25,26,29,35,38,40,42,43,52,55,69,79,83,84,87,93,96
+  tpcdsQueriesArg: supported
 
 jobs:
   beam_PostCommit_Java_Tpcds_Dataflow:

.github/workflows/beam_PostCommit_Java_Tpcds_Flink.yml

@@ -51,7 +51,7 @@ env:
     --dataSize=1GB
     --sourceType=PARQUET
     --dataDirectory=gs://beam-tpcds/datasets/parquet/nonpartitioned
-    --resultsDirectory=gs://beam-tpcds/results/flink/
+    --resultsDirectory=gs://temp-storage-for-perf-tests/tpcds/results/flink
     --tpcParallel=1
   INFLUXDB_USER: ${{ secrets.INFLUXDB_USER }}
   INFLUXDB_USER_PASSWORD: ${{ secrets.INFLUXDB_USER_PASSWORD }}
@@ -68,7 +68,7 @@ env:
     --baseInfluxMeasurement=tpcds 
     --exportSummaryToInfluxDB=true 
     --influxRetentionPolicy=forever 
-  tpcdsQueriesArg: 3,7,10,25,26,29,35,38,40,42,43,52,55,69,79,83,84,87,93,96
+  tpcdsQueriesArg: supported
 
 jobs:
   beam_PostCommit_Java_Tpcds_Flink:

.github/workflows/beam_PostCommit_Java_Tpcds_Spark.yml

@@ -49,7 +49,7 @@ env:
     --dataSize=1GB
     --sourceType=PARQUET
     --dataDirectory=gs://beam-tpcds/datasets/parquet/nonpartitioned
-    --resultsDirectory=gs://beam-tpcds/results/spark3-rdd/
+    --resultsDirectory=gs://temp-storage-for-perf-tests/tpcds/results/spark3-rdd
     --tpcParallel=1
   INFLUXDB_USER: ${{ secrets.INFLUXDB_USER }}
   INFLUXDB_USER_PASSWORD: ${{ secrets.INFLUXDB_USER_PASSWORD }}
@@ -66,7 +66,7 @@ env:
     --baseInfluxMeasurement=tpcds 
     --exportSummaryToInfluxDB=true 
     --influxRetentionPolicy=forever 
-  tpcdsQueriesArg: 3,7,10,25,26,29,35,38,40,42,43,52,55,69,79,83,84,87,93,96
+  tpcdsQueriesArg: supported
 
 jobs:
   beam_PostCommit_Java_Tpcds_Spark:

.github/workflows/beam_PostCommit_Python_Tpcds_Prism.yml

@@ -0,0 +1,75 @@
+# 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.
+
+name: PostCommit Python Tpcds Prism
+
+on:
+  schedule:
+    - cron: '30 4/6 * * *'
+  pull_request:
+    paths: ['release/trigger_all_tests.json', '.github/trigger_files/beam_PostCommit_Python_Tpcds_Prism.json']
+  workflow_dispatch:
+
+# Setting explicit permissions for the action
+permissions:
+  actions: write
+  pull-requests: read
+  checks: read
+  contents: read
+  deployments: read
+  id-token: none
+  issues: read
+  discussions: read
+  packages: read
+  pages: read
+  repository-projects: read
+  security-events: read
+  statuses: read
+
+# This allows a subsequently queued workflow run to interrupt previous runs
+concurrency:
+  group: '${{ github.workflow }} @ ${{ github.event.pull_request.number || github.sha || github.head_ref || github.ref }}-${{ github.event.schedule || github.event.sender.login }}'
+  cancel-in-progress: true
+
+env:
+  DEVELOCITY_ACCESS_KEY: ${{ secrets.BEAM_DEVELOCITY_ACCESS_KEY }}
+
+jobs:
+  beam_PostCommit_Python_Tpcds_Prism:
+    if: |
+      github.event_name == 'workflow_dispatch' ||
+      github.event_name == 'pull_request' ||
+      (github.event_name == 'schedule' && github.repository == 'apache/beam')
+    runs-on: [self-hosted, ubuntu-24.04, main]
+    timeout-minutes: 60
+    name: ${{ matrix.job_name }}
+    strategy:
+      matrix:
+        job_name: ["beam_PostCommit_Python_Tpcds_Prism"]
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          persist-credentials: false
+      - name: Setup repository
+        uses: ./.github/actions/setup-action
+      - name: Setup environment
+        uses: ./.github/actions/setup-environment-action
+        with:
+          python-version: default
+          go-version: default
+      - name: Run full TPC-DS suite on PrismRunner with regression check
+        uses: ./.github/actions/gradle-command-self-hosted-action
+        with:
+          gradle-command: :sdks:python:apache_beam:testing:benchmarks:tpcds:checkAllQueries

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/CalciteQueryPlanner.java

@@ -239,6 +239,11 @@ public BeamRelNode convertToBeamRel(String sqlStatement, QueryParameters queryPa
           String.format("Unable to convert query %s", sqlStatement), e);
     } catch (SqlParseException | ValidationException e) {
       throw new ParseException(String.format("Unable to parse query %s", sqlStatement), e);
+    } catch (RuntimeException e) {
+      if (e.getCause() instanceof SqlConversionException) {
+        throw new SqlConversionException(e.getMessage(), e.getCause());
+      }
+      throw e;
     } finally {
       planner.close();
     }

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/BeamRuleSets.java

@@ -26,6 +26,7 @@
 import org.apache.beam.sdk.extensions.sql.impl.rule.BeamCalcMergeRule;
 import org.apache.beam.sdk.extensions.sql.impl.rule.BeamCalcRule;
 import org.apache.beam.sdk.extensions.sql.impl.rule.BeamCoGBKJoinRule;
+import org.apache.beam.sdk.extensions.sql.impl.rule.BeamCrossJoinRule;
 import org.apache.beam.sdk.extensions.sql.impl.rule.BeamEnumerableConverterRule;
 import org.apache.beam.sdk.extensions.sql.impl.rule.BeamIOPushDownRule;
 import org.apache.beam.sdk.extensions.sql.impl.rule.BeamIntersectRule;
@@ -60,6 +61,7 @@ public class BeamRuleSets {
       ImmutableList.of(
           // Rules for window functions
           CoreRules.PROJECT_TO_LOGICAL_PROJECT_AND_WINDOW,
+          CoreRules.CALC_TO_WINDOW,
           // Rules so we only have to implement Calc
           CoreRules.FILTER_TO_CALC,
           CoreRules.PROJECT_TO_CALC,
@@ -144,6 +146,7 @@ public class BeamRuleSets {
           BeamUnnestRule.INSTANCE,
           BeamSideInputJoinRule.INSTANCE,
           BeamCoGBKJoinRule.INSTANCE,
+          BeamCrossJoinRule.INSTANCE,
           BeamSideInputLookupJoinRule.INSTANCE,
           BeamMatchRule.INSTANCE);
 

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/RelMdNodeStats.java

@@ -24,6 +24,7 @@
 import org.apache.beam.sdk.extensions.sql.impl.rel.BeamRelNode;
 import org.apache.beam.vendor.calcite.v1_40_0.com.google.common.collect.Table;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.RelNode;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.core.Values;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.metadata.MetadataDef;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.metadata.MetadataHandler;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider;
@@ -57,6 +58,11 @@ public NodeStats getNodeStats(RelNode rel, RelMetadataQuery mq) {
       return this.getBeamNodeStats((BeamRelNode) rel, bmq);
     }
 
+    if (rel instanceof Values) {
+      Values values = (Values) rel;
+      return NodeStats.create(values.getTuples().size());
+    }
+
     // We can later define custom methods for all different RelNodes to prevent hitting this point.
     // Similar to RelMdRowCount in calcite.
 

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCrossJoinRel.java

@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.sql.impl.rel;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionList;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.plan.RelOptCluster;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.plan.RelTraitSet;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.RelNode;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.core.CorrelationId;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.core.Join;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.core.JoinRelType;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rex.RexNode;
+
+/**
+ * A {@code BeamJoinRel} that implements a BOUNDED cartesian (CROSS) join, used for the literal-TRUE
+ * join conditions that Calcite emits when it decorrelates uncorrelated scalar / IN / EXISTS
+ * subqueries and lateral joins.
+ *
+ * <p>Only supported when BOTH inputs are bounded. The right side is broadcast as a side input and
+ * the left side is iterated; every (left, right) pair is emitted as one combined row (left values
+ * followed by right values).
+ *
+ * <p>First increment: INNER join with an always-true condition. A non-trivial (residual) condition
+ * is not yet evaluated here.
+ */
+public class BeamCrossJoinRel extends BeamJoinRel {
+
+  public BeamCrossJoinRel(
+      RelOptCluster cluster,
+      RelTraitSet traitSet,
+      RelNode left,
+      RelNode right,
+      RexNode condition,
+      Set<CorrelationId> variablesSet,
+      JoinRelType joinType) {
+    super(cluster, traitSet, left, right, condition, variablesSet, joinType);
+  }
+
+  @Override
+  public Join copy(
+      RelTraitSet traitSet,
+      RexNode conditionExpr,
+      RelNode left,
+      RelNode right,
+      JoinRelType joinType,
+      boolean semiJoinDone) {
+    return new BeamCrossJoinRel(
+        getCluster(), traitSet, left, right, conditionExpr, variablesSet, joinType);
+  }
+
+  @Override
+  public PTransform<PCollectionList<Row>, PCollection<Row>> buildPTransform() {
+    BeamRelNode leftRelNode = BeamSqlRelUtils.getBeamRelInput(left);
+    BeamRelNode rightRelNode = BeamSqlRelUtils.getBeamRelInput(right);
+    if (leftRelNode.isBounded() != PCollection.IsBounded.BOUNDED
+        || rightRelNode.isBounded() != PCollection.IsBounded.BOUNDED) {
+      throw new UnsupportedOperationException(
+          "Bounded cross join requires both inputs to be bounded.");
+    }
+    if (joinType != JoinRelType.INNER && joinType != JoinRelType.LEFT) {
+      throw new UnsupportedOperationException(
+          "Only INNER and LEFT cross join are supported for now, got " + joinType.name());
+    }
+    if (!condition.isAlwaysTrue()) {
+      throw new UnsupportedOperationException(
+          "Cross join with a residual (non-equi) condition is not supported yet.");
+    }
+    return new CrossJoin();
+  }
+
+  private class CrossJoin extends PTransform<PCollectionList<Row>, PCollection<Row>> {
+    @Override
+    public PCollection<Row> expand(PCollectionList<Row> pinput) {
+      PCollection<Row> leftRows =
+          pinput
+              .get(0)
+              .apply(
+                  "left_TimestampCombiner",
+                  Window.<Row>configure().withTimestampCombiner(TimestampCombiner.EARLIEST));
+      PCollection<Row> rightRows =
+          pinput
+              .get(1)
+              .apply(
+                  "right_TimestampCombiner",
+                  Window.<Row>configure().withTimestampCombiner(TimestampCombiner.EARLIEST));
+
+      final Schema outputSchema = CalciteUtils.toSchema(getRowType());
+      final PCollectionView<List<Row>> rightView = rightRows.apply(View.asList());
+      final boolean leftJoin = joinType == JoinRelType.LEFT;
+      final int rightFieldCount = getRowType().getFieldCount() - left.getRowType().getFieldCount();
+
+      return leftRows
+          .apply(
+              ParDo.of(new CrossJoinDoFn(outputSchema, rightView, leftJoin, rightFieldCount))
+                  .withSideInputs(rightView))
+          .setRowSchema(outputSchema);
+    }
+  }
+
+  /**
+   * Emits the cartesian product of each left row with the broadcast right rows. A static {@link
+   * DoFn} (not an anonymous/inner class) so it does not capture the non-serializable enclosing rel.
+   */
+  private static class CrossJoinDoFn extends DoFn<Row, Row> {
+    private final Schema outputSchema;
+    private final PCollectionView<List<Row>> rightView;
+    private final boolean leftJoin;
+    private final int rightFieldCount;
+
+    CrossJoinDoFn(
+        Schema outputSchema,
+        PCollectionView<List<Row>> rightView,
+        boolean leftJoin,
+        int rightFieldCount) {
+      this.outputSchema = outputSchema;
+      this.rightView = rightView;
+      this.leftJoin = leftJoin;
+      this.rightFieldCount = rightFieldCount;
+    }
+
+    // Row.Builder.addValues expects List<@Nullable Object>, while Row.getBaseValues() returns
+    // List<@NonNull Object>, so it is safe to suppress nullness here.
+    @SuppressWarnings("nullness")
+    @ProcessElement
+    public void processElement(ProcessContext c) {
+      Row leftRow = c.element();
+      List<Row> rightRowList = c.sideInput(rightView);
+      for (Row rightRow : rightRowList) {
+        c.output(
+            Row.withSchema(outputSchema)
+                .addValues(leftRow.getBaseValues())
+                .addValues(rightRow.getBaseValues())
+                .build());
+      }
+      if (leftJoin && rightRowList.isEmpty()) {
+        c.output(
+            Row.withSchema(outputSchema)
+                .addValues(leftRow.getBaseValues())
+                .addValues(Collections.nCopies(rightFieldCount, null))
+                .build());
+      }
+    }
+  }
+}