From 294d6c92f13838da254b23ca92a4332a408295a7 Mon Sep 17 00:00:00 2001 From: Dhruv Dagar Date: Fri, 11 Sep 2026 13:12:28 +0530 Subject: [PATCH 1/4] [Go SDK] Add portable logical type abstraction --- .../graphx/schema/portable_logicaltypes.go | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 sdks/go/pkg/beam/core/runtime/graphx/schema/portable_logicaltypes.go diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/portable_logicaltypes.go b/sdks/go/pkg/beam/core/runtime/graphx/schema/portable_logicaltypes.go new file mode 100644 index 000000000000..27f2643ebc67 --- /dev/null +++ b/sdks/go/pkg/beam/core/runtime/graphx/schema/portable_logicaltypes.go @@ -0,0 +1,45 @@ +// 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 schema + +import ( + "reflect" + + pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1" +) + +// PortableLogicalType describes a logical type that can be represented in a +// portable Beam schema. It is intentionally separate from LogicalType for now: +// LogicalType is the existing Go schema-registration abstraction, while this +// interface models the wire-level contract needed for cross-SDK schemas. +type PortableLogicalType interface { + // URN returns the portable Beam logical type URN. + URN() string + + // GoType returns the Go type represented by this logical type. + GoType() reflect.Type + + // Representation returns the underlying Beam field type used on the wire. + Representation() *pipepb.FieldType + + // ArgumentType returns the Beam field type used for the logical type + // argument, or nil for non-parameterized logical types. + ArgumentType() *pipepb.FieldType + + // Argument returns the argument value, or nil for non-parameterized logical + // types. + Argument() *pipepb.FieldValue +} From 4d82ecba3b74a7f98d5e46271c434fa69b6d3f3d Mon Sep 17 00:00:00 2001 From: Dhruv Dagar Date: Sun, 4 Oct 2026 09:24:43 +0530 Subject: [PATCH 2/4] Fix Spark MapState and SetState emptiness after removals --- .../runners/spark/stateful/SparkStateInternals.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java index 51ceb4c87309..4f744ab3ab12 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java @@ -432,7 +432,11 @@ private Map readAsMap() { public void remove(MapKeyT key) { Map sparkMapState = readAsMap(); sparkMapState.remove(key); - writeValue(sparkMapState); + if (sparkMapState.isEmpty()) { + clear(); + } else { + writeValue(sparkMapState); + } } @Override @@ -537,7 +541,11 @@ public ReadableState addIfAbsent(InputT input) { public void remove(InputT input) { Set sparkSetState = readAsSet(); sparkSetState.remove(input); - writeValue(sparkSetState); + if (sparkSetState.isEmpty()) { + clear(); + } else { + writeValue(sparkSetState); + } } @Override From d4a3aedbfe1d993f25a5b482b929541c7997a640 Mon Sep 17 00:00:00 2001 From: Dhruv Dagar Date: Sun, 4 Oct 2026 09:24:45 +0530 Subject: [PATCH 3/4] Add regression tests for state emptiness after removal --- .../apache/beam/runners/core/StateInternalsTest.java | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/StateInternalsTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/StateInternalsTest.java index e15249969f26..6988a492736a 100644 --- a/runners/core-java/src/test/java/org/apache/beam/runners/core/StateInternalsTest.java +++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/StateInternalsTest.java @@ -225,6 +225,11 @@ public void testSet() throws Exception { assertThat(later.read(), hasItems("C", "D")); assertFalse(later.contains("A").read()); + value.remove("B"); + value.remove("C"); + value.remove("D"); + assertTrue(value.isEmpty().read()); + // clear value.clear(); assertThat(value.read(), Matchers.emptyIterable()); @@ -389,6 +394,11 @@ public void testMap() throws Exception { // isEmpty assertFalse(value.isEmpty().read()); + value.remove("B"); + value.remove("D"); + value.remove("E"); + assertTrue(value.isEmpty().read()); + // clear value.clear(); assertThat(value.entries().read(), Matchers.emptyIterable()); From 4b466bd41c6431498aff9bbe2eba7bb41dd50065 Mon Sep 17 00:00:00 2001 From: Dhruv Dagar Date: Sun, 4 Oct 2026 09:37:19 +0530 Subject: [PATCH 4/4] Remove unrelated upstream file from BEAM-40264 branch --- .../graphx/schema/portable_logicaltypes.go | 45 ------------------- 1 file changed, 45 deletions(-) delete mode 100644 sdks/go/pkg/beam/core/runtime/graphx/schema/portable_logicaltypes.go diff --git a/sdks/go/pkg/beam/core/runtime/graphx/schema/portable_logicaltypes.go b/sdks/go/pkg/beam/core/runtime/graphx/schema/portable_logicaltypes.go deleted file mode 100644 index 27f2643ebc67..000000000000 --- a/sdks/go/pkg/beam/core/runtime/graphx/schema/portable_logicaltypes.go +++ /dev/null @@ -1,45 +0,0 @@ -// 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 schema - -import ( - "reflect" - - pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1" -) - -// PortableLogicalType describes a logical type that can be represented in a -// portable Beam schema. It is intentionally separate from LogicalType for now: -// LogicalType is the existing Go schema-registration abstraction, while this -// interface models the wire-level contract needed for cross-SDK schemas. -type PortableLogicalType interface { - // URN returns the portable Beam logical type URN. - URN() string - - // GoType returns the Go type represented by this logical type. - GoType() reflect.Type - - // Representation returns the underlying Beam field type used on the wire. - Representation() *pipepb.FieldType - - // ArgumentType returns the Beam field type used for the logical type - // argument, or nil for non-parameterized logical types. - ArgumentType() *pipepb.FieldType - - // Argument returns the argument value, or nil for non-parameterized logical - // types. - Argument() *pipepb.FieldValue -}