[pull] master from apache:master

.agent/skills/README.md

@@ -25,7 +25,9 @@ This directory contains skills that help the agent perform specialized tasks in
 
 | Skill | Description |
 |-------|-------------|
+| [adding-new-metadata](adding-new-metadata/SKILL.md) | Guide on how to add and propagate new metadata fields in WindowedValue to avoid metadata loss |
 | [beam-concepts](beam-concepts/SKILL.md) | Core Beam programming model (PCollections, PTransforms, windowing, triggers) |
+| [beam-dofn-modernizer](beam-dofn-modernizer/SKILL.md) | Rewrite Apache Beam DoFn methods to remove legacy ProcessContext/OnTimerContext |
 | [ci-cd](ci-cd/SKILL.md) | GitHub Actions workflows, debugging CI failures, triggering tests |
 | [contributing](contributing/SKILL.md) | PR workflow, issue management, code review, release cycles |
 | [gradle-build](gradle-build/SKILL.md) | Build commands, flags, publishing, troubleshooting |
@@ -35,6 +37,7 @@ This directory contains skills that help the agent perform specialized tasks in
 | [license-compliance](license-compliance/SKILL.md) | Apache 2.0 license headers for all new files |
 | [python-development](python-development/SKILL.md) | Python SDK environment setup, testing, building pipelines |
 | [runners](runners/SKILL.md) | Direct, Dataflow, Flink, Spark runner configuration |
+| [yaml-development](yaml-development/SKILL.md) | YAML SDK development, environment setup, testing, and key concepts |
 
 ## How Skills Work
 

.github/trigger_files/beam_PostCommit_Go.json

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

CHANGES.md

@@ -59,10 +59,10 @@
 
 ## Highlights
 
-* New highly anticipated feature X added to Python SDK ([#X](https://github.com/apache/beam/issues/X)).
-* New highly anticipated feature Y added to Java SDK ([#Y](https://github.com/apache/beam/issues/Y)).
+* Python SDK now supports memory profiling with Memray ([#38853](https://github.com/apache/beam/issues/38853)).
 * (Python) Added [Qdrant](https://qdrant.tech/) VectorDatabaseWriteConfig implementation ([#38141](https://github.com/apache/beam/issues/38141)).
 
+
 ## I/Os
 
 * Support for reading from Delta Lake added (Java) ([#38551](https://github.com/apache/beam/issues/38551)).
@@ -71,10 +71,10 @@
 ## New Features / Improvements
 
 * (Java) Enabled state tag encoding v2 by default for new Dataflow Streaming Engine jobs. It can be disabled by passing `--experiments=disable_streaming_engine_state_tag_encoding_v2` or `--updateCompatibilityVersion=2.74.0` pipeline option. Note that the tag encoding version cannot change during a job update. Jobs using tag encoding v2 (enabled by default for new jobs on 2.75.0+) cannot be downgraded to Beam versions prior to 2.73.0, as only versions 2.73.0 and later support tag encoding v2. ([#38705](https://github.com/apache/beam/issues/38705)).
+* (Python) Added instrumentation to support off-the-shelf profiling agents when launching Python SDK Harness ([#38853](https://github.com/apache/beam/issues/38853)).
 
 ## Breaking Changes
 
-* X behavior was changed ([#X](https://github.com/apache/beam/issues/X)).
 * (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.
@@ -85,7 +85,8 @@
 
 ## Bugfixes
 
-* Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* Fixed GCS filesystem glob matching to correctly handle `/` in object names and support `**` for recursive matching (Go) ([#38059](https://github.com/apache/beam/issues/38059)).
+* Fixed BigQueryEnrichmentHandler batch mode dropping earlier requests when multiple requests share the same enrichment key (Python) ([#38035](https://github.com/apache/beam/issues/38035)).
 * Fixed IcebergIO writing manifest column bounds padded with trailing `0x00` bytes, which broke equality predicate pushdown in some query engines (Java) ([#38580](https://github.com/apache/beam/issues/38580)).
 
 ## Security Fixes

infra/iam/users.yml

@@ -486,6 +486,12 @@
   - role: roles/cloudfunctions.invoker
   - role: roles/iam.serviceAccountTokenCreator
   - role: roles/storage.objectViewer
+- username: hansmarcus
+  email: hansmarcus14@gmail.com
+  member_type: user
+  permissions:
+  - role: projects/apache-beam-testing/roles/beam_viewer
+  - role: projects/apache-beam-testing/roles/beam_writer
 - username: harrisonlim
   email: harrisonlim@google.com
   member_type: user
@@ -1241,4 +1247,4 @@
   member_type: user
   permissions:
   - role: projects/apache-beam-testing/roles/beam_viewer
-  - role: projects/apache-beam-testing/roles/beam_writer
+  - role: projects/apache-beam-testing/roles/beam_writer

sdks/go/pkg/beam/io/filesystem/gcs/gcs.go

@@ -21,7 +21,8 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"path/filepath"
+	"regexp"
+	"strings"
 	"time"
 
 	"cloud.google.com/go/storage"
@@ -38,6 +39,76 @@ const (
 	projectBillingHook = "beam:go:hook:filesystem:billingproject"
 )
 
+// globToRegex translates a glob pattern to a regular expression.
+// It differs from filepath.Match in that:
+//   - / is treated as a regular character (not a separator), since GCS object
+//     names are flat with / being just another character
+//   - ** matches any sequence of characters including / (zero or more)
+//   - **/  matches zero or more path segments (e.g., "" or "dir/" or "dir/subdir/")
+//   - * matches any sequence of characters except / (zero or more)
+//   - ? matches any single character except /
+//
+// This matches the behavior of the Python and Java SDKs.
+func globToRegex(pattern string) (*regexp.Regexp, error) {
+	var result strings.Builder
+	result.WriteString("^")
+
+	for i := 0; i < len(pattern); i++ {
+		c := pattern[i]
+		switch c {
+		case '*':
+			// Check for ** (double asterisk)
+			if i+1 < len(pattern) && pattern[i+1] == '*' {
+				// Check if followed by / (e.g., "**/" matches zero or more path segments)
+				if i+2 < len(pattern) && pattern[i+2] == '/' {
+					// **/ matches "" or "something/" or "a/b/c/"
+					result.WriteString("(?:.*/)?")
+					i += 2 // Skip the second * and the /
+				} else {
+					// ** at end or before non-slash matches any characters
+					result.WriteString(".*")
+					i++ // Skip the second *
+				}
+			} else {
+				result.WriteString("[^/]*")
+			}
+		case '?':
+			result.WriteString("[^/]")
+		case '[':
+			// Character class - find the closing bracket
+			j := i + 1
+			if j < len(pattern) && pattern[j] == '!' {
+				j++
+			}
+			if j < len(pattern) && pattern[j] == ']' {
+				j++
+			}
+			for j < len(pattern) && pattern[j] != ']' {
+				j++
+			}
+			if j >= len(pattern) {
+				return nil, fmt.Errorf("syntax error: unclosed '[' in pattern %q", pattern)
+			} else {
+				// Copy the character class, converting ! to ^ for negation
+				result.WriteByte('[')
+				content := pattern[i+1 : j]
+				if len(content) > 0 && content[0] == '!' {
+					result.WriteByte('^')
+					content = content[1:]
+				}
+				result.WriteString(content)
+				result.WriteByte(']')
+				i = j
+			}
+		default:
+			result.WriteString(regexp.QuoteMeta(string(c)))
+		}
+	}
+
+	result.WriteString("$") // match end
+	return regexp.Compile(result.String())
+}
+
 var billingProject string = ""
 
 func init() {
@@ -107,6 +178,15 @@ func (f *fs) List(ctx context.Context, glob string) ([]string, error) {
 		return nil, err
 	}
 
+	// Compile the glob pattern to a regex. We use a custom glob-to-regex
+	// translation that treats / as a regular character (not a separator),
+	// since GCS object names are flat. This also supports ** for recursive
+	// matching, similar to the Java and Python SDKs.
+	re, err := globToRegex(object)
+	if err != nil {
+		return nil, fmt.Errorf("invalid glob pattern %q: %w", object, err)
+	}
+
 	var candidates []string
 
 	// We handle globs by list all candidates and matching them here.
@@ -125,11 +205,7 @@ func (f *fs) List(ctx context.Context, glob string) ([]string, error) {
 			return nil, err
 		}
 
-		match, err := filepath.Match(object, obj.Name)
-		if err != nil {
-			return nil, err
-		}
-		if match {
+		if re.MatchString(obj.Name) {
 			candidates = append(candidates, obj.Name)
 		}
 	}

sdks/go/pkg/beam/io/filesystem/gcs/gcs_test.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"io"
 	"sort"
+	"strings"
 	"testing"
 	"time"
 
@@ -271,6 +272,151 @@ func TestGCS_copy(t *testing.T) {
 	}
 }
 
+func TestGlobToRegex(t *testing.T) {
+	tests := []struct {
+		pattern string
+		name    string
+		want    bool
+	}{
+		// Single * should NOT match / in object names
+		{"*.txt", "file.txt", true},
+		{"*.txt", "dir/file.txt", false},
+		{"prefix*", "prefix123", true},
+		{"prefix*", "prefix/subdir", false},
+
+		// ** should match any characters including /
+		{"**", "file.txt", true},
+		{"**", "dir/file.txt", true},
+		{"**", "dir/subdir/file.txt", true},
+		{"prefix/**", "prefix/file.txt", true},
+		{"prefix/**", "prefix/subdir/file.txt", true},
+		{"**/file.txt", "file.txt", true},
+		{"**/file.txt", "dir/file.txt", true},
+		{"**/file.txt", "dir/subdir/file.txt", true},
+
+		// Mixed patterns
+		{"dir/*.txt", "dir/file.txt", true},
+		{"dir/*.txt", "dir/subdir/file.txt", false},
+		{"dir/**/*.txt", "dir/file.txt", true},
+		{"dir/**/*.txt", "dir/subdir/file.txt", true},
+		{"dir/**/file.txt", "dir/file.txt", true},
+		{"dir/**/file.txt", "dir/a/b/c/file.txt", true},
+
+		// ? should match any single character except /
+		{"file?.txt", "file1.txt", true},
+		{"file?.txt", "file12.txt", false},
+		{"file?.txt", "file/.txt", false}, // ? should not cross /
+		{"dir?file.txt", "dir/file.txt", false},
+
+		// Character classes
+		{"file[0-9].txt", "file1.txt", true},
+		{"file[0-9].txt", "filea.txt", false},
+		{"file[!0-9].txt", "filea.txt", true},
+		{"file[!0-9].txt", "file1.txt", false},
+
+		// Exact match (no wildcards)
+		{"exact.txt", "exact.txt", true},
+		{"exact.txt", "notexact.txt", false},
+
+		// Regex special characters should be escaped
+		{"file.txt", "file.txt", true},
+		{"file.txt", "fileXtxt", false},
+		{"file(1).txt", "file(1).txt", true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.pattern+"_"+tt.name, func(t *testing.T) {
+			re, err := globToRegex(tt.pattern)
+			if err != nil {
+				t.Fatalf("globToRegex(%q) error = %v", tt.pattern, err)
+			}
+			got := re.MatchString(tt.name)
+			if got != tt.want {
+				t.Errorf("globToRegex(%q).MatchString(%q) = %v, want %v", tt.pattern, tt.name, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestGlobToRegex_errors(t *testing.T) {
+	tests := []struct {
+		pattern string
+		wantErr string
+	}{
+		{"file[abc.txt", "unclosed '['"},
+		{"[invalid", "unclosed '['"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.pattern, func(t *testing.T) {
+			_, err := globToRegex(tt.pattern)
+			if err == nil {
+				t.Errorf("globToRegex(%q) expected error containing %q, got nil", tt.pattern, tt.wantErr)
+			} else if !strings.Contains(err.Error(), tt.wantErr) {
+				t.Errorf("globToRegex(%q) error = %v, want error containing %q", tt.pattern, err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestGCS_listWithSlashesInObjectNames(t *testing.T) {
+	ctx := context.Background()
+	bucket := "beamgogcsfilesystemtest"
+	dirPath := "gs://" + bucket
+
+	// Create server with objects that have / in their names
+	server := fakestorage.NewServer([]fakestorage.Object{
+		{ObjectAttrs: fakestorage.ObjectAttrs{BucketName: bucket, Name: "file.txt"}, Content: []byte("")},
+		{ObjectAttrs: fakestorage.ObjectAttrs{BucketName: bucket, Name: "dir/file.txt"}, Content: []byte("")},
+		{ObjectAttrs: fakestorage.ObjectAttrs{BucketName: bucket, Name: "dir/subdir/file.txt"}, Content: []byte("")},
+		{ObjectAttrs: fakestorage.ObjectAttrs{BucketName: bucket, Name: "other.txt"}, Content: []byte("")},
+	})
+	t.Cleanup(server.Stop)
+	c := &fs{client: server.Client()}
+
+	tests := []struct {
+		glob string
+		want []string
+	}{
+		// Single * should only match top-level files
+		{dirPath + "/*.txt", []string{dirPath + "/file.txt", dirPath + "/other.txt"}},
+		// ** should match all files recursively
+		{dirPath + "/**", []string{
+			dirPath + "/file.txt",
+			dirPath + "/dir/file.txt",
+			dirPath + "/dir/subdir/file.txt",
+			dirPath + "/other.txt",
+		}},
+		// dir/* should only match immediate children
+		{dirPath + "/dir/*", []string{dirPath + "/dir/file.txt"}},
+		// dir/** should match all descendants
+		{dirPath + "/dir/**", []string{
+			dirPath + "/dir/file.txt",
+			dirPath + "/dir/subdir/file.txt",
+		}},
+		// Deeply nested ** matching (core scenario from issue #38059)
+		{dirPath + "/dir/subdir/**", []string{
+			dirPath + "/dir/subdir/file.txt",
+		}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.glob, func(t *testing.T) {
+			got, err := c.List(ctx, tt.glob)
+			if err != nil {
+				t.Fatalf("List(%q) error = %v", tt.glob, err)
+			}
+
+			sort.Strings(got)
+			sort.Strings(tt.want)
+
+			if !cmp.Equal(got, tt.want) {
+				t.Errorf("List(%q) = %v, want %v", tt.glob, got, tt.want)
+			}
+		})
+	}
+}
+
 func createFakeGCSServer(tb testing.TB) *fakestorage.Server {
 	tb.Helper()
 

sdks/java/extensions/sql/src/main/codegen/includes/parserImpls.ftl

@@ -381,7 +381,7 @@ SqlCreate SqlCreateDatabase(Span s, boolean replace) :
 }
 
 /**
- * USE DATABASE ( catalog_name '.' )? database_name
+ * USE [ DATABASE ] ( catalog_name '.' )? database_name
  */
 SqlCall SqlUseDatabase(Span s, String scope) :
 {
@@ -391,7 +391,7 @@ SqlCall SqlUseDatabase(Span s, String scope) :
     <USE> {
         s.add(this);
     }
-    <DATABASE>
+    [ <DATABASE> ]
     databaseName = CompoundIdentifier()
     {
         return new SqlUseDatabase(

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

@@ -55,7 +55,7 @@ public static SqlNode column(
   }
 
   /** Returns the schema in which to create an object. */
-  static Pair<CalciteSchema, String> schema(
+  public static Pair<CalciteSchema, String> schema(
       CalcitePrepare.Context context, boolean mutable, SqlIdentifier id) {
     CalciteSchema rootSchema = mutable ? context.getMutableRootSchema() : context.getRootSchema();
     @Nullable CalciteSchema schema = null;
@@ -72,7 +72,10 @@ static Pair<CalciteSchema, String> schema(
     return Pair.of(checkStateNotNull(schema, "Got null sub-schema for path '%s'", path), name(id));
   }
 
-  private static @Nullable CalciteSchema childSchema(CalciteSchema rootSchema, List<String> path) {
+  public static @Nullable CalciteSchema childSchema(CalciteSchema rootSchema, List<String> path) {
+    if (path == null) {
+      return null;
+    }
     @Nullable CalciteSchema schema = rootSchema;
     for (String p : path) {
       if (schema == null) {