apache · kennknowles · Aug 27, 2026
diff --git a/CHANGES.md b/CHANGES.md
@@ -183,6 +183,15 @@
 
 ## Breaking Changes
 
+* (Java) Beam SQL now names the transforms a query expands into after the operations fused into them
+  (`Filter;Project`) instead of after the relational node class and a JVM-global counter
+  (`BeamCalcRel_57`). The old counter was JVM-global, so the old names differed between runs of the
+  same query. The new names do not, but two stages with the same provenance are still told apart by
+  an occurrence suffix (`Filter;Project #2`) numbered within the query, so reshaping a plan can
+  renumber the stages of that query. Since Dataflow matches streaming pipelines for update by step
+  name, a running streaming pipeline must either be drained or be started with
+  `--experiments=legacy-sql-transform-names` to keep the old names. That experiment exists only to
+  carry running pipelines over this change and is expected to be removed two releases from now.
 * (Python) Typehints of dataclass fields are honored during type inferences. To restore the behavior of fallback-to-any,
   use pipeline option `--exclude_infer_dataclass_field_type` ([#38797](https://github.com/apache/beam/issues/38797)).
   However fixing forward is recommended.

diff --git a/...nsions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/CalciteQueryPlanner.java b/...nsions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/CalciteQueryPlanner.java
@@ -30,6 +30,8 @@
 import org.apache.beam.sdk.extensions.sql.impl.rel.BeamLogicalConvention;
 import org.apache.beam.sdk.extensions.sql.impl.rel.BeamRelNode;
 import org.apache.beam.sdk.extensions.sql.impl.rel.BeamSqlRelUtils;
+import org.apache.beam.sdk.extensions.sql.impl.rel.StageName;
+import org.apache.beam.sdk.extensions.sql.impl.rule.StageNameRule;
 import org.apache.beam.sdk.extensions.sql.impl.udf.BeamBuiltinFunctionProvider;
 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.config.CalciteConnectionConfig;
@@ -155,7 +157,9 @@ public FrameworkConfig defaultConfig(JdbcConnection connection, Collection<RuleS
         .defaultSchema(defaultSchema)
         .traitDefs(traitDefs)
         .context(Contexts.of(connection.config()))
-        .ruleSets(ruleSets.toArray(new RuleSet[0]))
+        // Wrapped here rather than in BeamRuleSets so that callers assembling their own rule set
+        // still see the rules by their own types.
+        .ruleSets(StageNameRule.wrapAll(ruleSets).toArray(new RuleSet[0]))
         .costFactory(BeamCostModel.FACTORY)
         .typeSystem(connection.getTypeFactory().getTypeSystem())
         .operatorTable(
@@ -211,7 +215,10 @@ public BeamRelNode convertToBeamRel(String sqlStatement, QueryParameters queryPa
                 relNode,
                 new ParameterBinder(root.rel.getCluster().getRexBuilder(), queryParameters));
       }
-      LOG.info("SQLPlan>\n{}", BeamSqlRelUtils.explainLazily(root.rel));
+      StageName.register(relNode.getCluster());
+      // Give every node a name to be composed from before any rule fuses it away.
+      relNode = StageName.backfill(relNode);
+      LOG.info("SQLPlan>\n{}", BeamSqlRelUtils.explainLazily(relNode));
       RelTraitSet desiredTraits =
           relNode
               .getTraitSet()

diff --git a/...java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/JdbcDriver.java b/...java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/JdbcDriver.java
@@ -30,6 +30,7 @@
 import java.util.function.Consumer;
 import org.apache.beam.sdk.extensions.sql.SqlTransform;
 import org.apache.beam.sdk.extensions.sql.impl.planner.BeamRuleSets;
+import org.apache.beam.sdk.extensions.sql.impl.rule.StageNameRule;
 import org.apache.beam.sdk.extensions.sql.meta.catalog.CatalogManager;
 import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
 import org.apache.beam.sdk.options.PipelineOptions;
@@ -80,7 +81,7 @@ public class JdbcDriver extends Driver {
             planner -> {
               for (RuleSet ruleSet : BeamRuleSets.getRuleSets()) {
                 for (RelOptRule rule : ruleSet) {
-                  planner.addRule(rule);
+                  StageNameRule.addTo(planner, rule);
                 }
               }
               planner.removeRule(CoreRules.CALC_REMOVE);

diff --git a/...ns/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/AbstractBeamCalcRel.java b/...ns/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/AbstractBeamCalcRel.java
@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.extensions.sql.impl.rel;
 
+import java.util.List;
 import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.extensions.sql.impl.planner.BeamCostModel;
 import org.apache.beam.sdk.extensions.sql.impl.planner.BeamRelMetadataQuery;
@@ -26,10 +27,12 @@
 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.Calc;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.hint.RelHint;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.metadata.RelMetadataQuery;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rex.RexLocalRef;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rex.RexNode;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rex.RexProgram;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 
 /** BeamRelNode to replace {@code Project} and {@code Filter} node. */
 @Internal
@@ -40,7 +43,16 @@ public abstract class AbstractBeamCalcRel extends Calc implements BeamRelNode {
 
   public AbstractBeamCalcRel(
       RelOptCluster cluster, RelTraitSet traits, RelNode input, RexProgram program) {
-    super(cluster, traits, input, program);
+    this(cluster, traits, ImmutableList.of(), input, program);
+  }
+
+  public AbstractBeamCalcRel(
+      RelOptCluster cluster,
+      RelTraitSet traits,
+      List<RelHint> hints,
+      RelNode input,
+      RexProgram program) {
+    super(cluster, traits, hints, input, program);
   }
 
   public boolean isInputSortRelAndLimitOnly() {

diff --git a/...extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java b/...extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
@@ -90,6 +90,7 @@
 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.Calc;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.hint.RelHint;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rel.metadata.RelMetadataQuery;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rex.RexBuilder;
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rex.RexCall;
@@ -151,12 +152,26 @@ public static long timestampToCalciteMillis(java.time.Instant instant) {
   }
 
   public BeamCalcRel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RexProgram program) {
-    super(cluster, traits, input, program);
+    this(cluster, traits, ImmutableList.of(), input, program);
+  }
+
+  public BeamCalcRel(
+      RelOptCluster cluster,
+      RelTraitSet traits,
+      List<RelHint> hints,
+      RelNode input,
+      RexProgram program) {
+    super(cluster, traits, hints, input, program);
   }
 
   @Override
   public Calc copy(RelTraitSet traitSet, RelNode input, RexProgram program) {
-    return new BeamCalcRel(getCluster(), traitSet, input, program);
+    return new BeamCalcRel(getCluster(), traitSet, hints, input, program);
+  }
+
+  @Override
+  public RelNode withHints(List<RelHint> hintList) {
+    return new BeamCalcRel(getCluster(), traitSet, hintList, input, program);
   }
 
   @Override

diff --git a/...nsions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSqlRelUtils.java b/...nsions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSqlRelUtils.java
@@ -24,6 +24,7 @@
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.extensions.sql.impl.planner.BeamRelMetadataQuery;
 import org.apache.beam.sdk.extensions.sql.impl.planner.NodeStats;
+import org.apache.beam.sdk.options.ExperimentalOptions;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.util.Preconditions;
@@ -46,22 +47,23 @@ public class BeamSqlRelUtils {
   public static final String ERROR = "error";
 
   public static PCollection<Row> toPCollection(Pipeline pipeline, BeamRelNode node) {
-    return toPCollection(pipeline, node, null, new HashMap());
+    return toPCollection(pipeline, node, null, new HashMap(), new HashMap<>());
   }
 
   public static PCollection<Row> toPCollection(
       Pipeline pipeline,
       BeamRelNode node,
       @Nullable PTransform<PCollection<Row>, ? extends POutput> errorTransformer) {
-    return toPCollection(pipeline, node, errorTransformer, new HashMap());
+    return toPCollection(pipeline, node, errorTransformer, new HashMap(), new HashMap<>());
   }
 
   /** Transforms the inputs into a PInput. */
   private static PCollectionList<Row> buildPCollectionList(
       List<RelNode> inputRels,
       Pipeline pipeline,
       @Nullable PTransform<PCollection<Row>, ? extends POutput> errorTransformer,
-      Map<Integer, PCollection<Row>> cache) {
+      Map<Integer, PCollection<Row>> cache,
+      Map<String, Integer> usedNames) {
     if (inputRels.isEmpty()) {
       return PCollectionList.empty(pipeline);
     } else {
@@ -79,7 +81,7 @@ private static PCollectionList<Row> buildPCollectionList(
                       beamRel = (BeamRelNode) input;
                     }
                     return BeamSqlRelUtils.toPCollection(
-                        pipeline, beamRel, errorTransformer, cache);
+                        pipeline, beamRel, errorTransformer, cache, usedNames);
                   })
               .collect(Collectors.toList()));
     }
@@ -93,15 +95,17 @@ static PCollection<Row> toPCollection(
       Pipeline pipeline,
       BeamRelNode node,
       @Nullable PTransform<PCollection<Row>, ? extends POutput> errorTransformer,
-      Map<Integer, PCollection<Row>> cache) {
+      Map<Integer, PCollection<Row>> cache,
+      Map<String, Integer> usedNames) {
     PCollection<Row> output = cache.get(node.getId());
     if (output != null) {
       return output;
     }
 
-    String name = node.getClass().getSimpleName() + "_" + node.getId();
+    String name = uniqueName(usedNames, transformName(pipeline, node));
     PCollectionList<Row> input =
-        buildPCollectionList(node.getPCollectionInputs(), pipeline, errorTransformer, cache);
+        buildPCollectionList(
+            node.getPCollectionInputs(), pipeline, errorTransformer, cache, usedNames);
     PTransform<PCollectionList<Row>, PCollection<Row>> transform =
         node.buildPTransform(errorTransformer);
     output = Pipeline.applyTransform(name, input, transform);
@@ -110,6 +114,35 @@ static PCollection<Row> toPCollection(
     return output;
   }
 
+  /**
+   * Names the composite that {@code node} expands into after the stage it was composed from, or
+   * after the node's own type when nothing composed it.
+   */
+  private static String transformName(Pipeline pipeline, BeamRelNode node) {
+    if (ExperimentalOptions.hasExperiment(pipeline.getOptions(), StageName.LEGACY_EXPERIMENT)) {
+      return node.getClass().getSimpleName() + "_" + node.getId();
+    }
+    String label = StageName.renderedName(node);
+    return label == null || label.isEmpty() ? node.getClass().getSimpleName() : label;
+  }
+
+  /**
+   * Disambiguates repeated names in DFS order, within the plan being expanded.
+   *
+   * <p>Names no longer carry a rel id, so a plan can legitimately contain two stages with the same
+   * provenance. {@link Pipeline} would uniquify them itself, but then reports the pipeline as not
+   * having stable unique names, which is fatal under {@code --stableUniqueNames=ERROR}.
+   *
+   * <p>Counting per plan rather than per pipeline is what makes the numbering stable: every caller
+   * expands a plan either into a fresh pipeline or inside {@code SqlTransform}'s own composite, so
+   * names only have to be unique among the stages of one query. A counter shared across a pipeline
+   * would let an unrelated query added elsewhere renumber stages that did not themselves change.
+   */
+  private static String uniqueName(Map<String, Integer> usedNames, String name) {
+    int occurrence = usedNames.merge(name, 1, Integer::sum);
+    return occurrence == 1 ? name : name + " #" + occurrence;
+  }
+
   public static BeamRelNode getBeamRelInput(RelNode input) {
     if (input instanceof RelSubset) {
       // go with known best input