Add filter() method to PCollection in TypeScript SDK (#37408)

- Add filter() method following same pattern as map() and flatMap()
- Add 5 test cases covering various filtering scenarios
- Support optional context parameter for side inputs

Author: junaiddshaukat

sdks/typescript/src/apache_beam/pvalue.ts

@@ -163,6 +163,55 @@ export class PCollection<T> {
     );
   }
 
+  /**
+   * Returns a PCollection containing only elements that satisfy the given
+   * predicate function.
+   *
+   * This is analogous to JavaScript's `Array.filter()` method.
+   *
+   * Example usage:
+   * ```
+   * const evens = pcoll.filter(x => x % 2 === 0);
+   * const positives = pcoll.filter(x => x > 0);
+   * ```
+   *
+   * @param fn A predicate function that returns true for elements to keep,
+   *           false for elements to filter out. The function receives the
+   *           element and optionally a context object.
+   * @param context Optional context object to pass to the predicate function.
+   * @returns A new PCollection containing only the elements for which the
+   *          predicate returned true.
+   */
+  filter<ContextT extends Object | undefined = undefined>(
+    fn:
+      | (ContextT extends undefined ? (element: T) => boolean : never)
+      | ((element: T, context: ContextT) => boolean),
+    context: ContextT = undefined!,
+  ): PCollection<T> {
+    if (extractContext(fn)) {
+      context = { ...extractContext(fn), ...context };
+    }
+    return this.apply(
+      withName(
+        "filter(" + extractName(fn) + ")",
+        parDo<T, T, ContextT>(
+          {
+            process: function (element: T, context: ContextT) {
+              // Return the element wrapped in an array if predicate is true,
+              // otherwise return an empty array to filter it out.
+              const keep =
+                context === null || context === undefined
+                  ? (fn as (element: T) => boolean)(element)
+                  : fn(element, context);
+              return keep ? [element] : [];
+            },
+          },
+          context,
+        ),
+      ),
+    );
+  }
+
   root(): Root {
     return new Root(this.pipeline);
   }

sdks/typescript/test/primitives_test.ts

@@ -96,6 +96,51 @@ export function suite(runner: beam.Runner = directRunner()) {
       });
     });
 
+    it("runs a filter", async function () {
+      await runner.run((root) => {
+        root
+          .apply(beam.create([1, 2, 3, 4, 5, 6]))
+          .filter((x) => x % 2 === 0)
+          .apply(testing.assertDeepEqual([2, 4, 6]));
+      });
+    });
+
+    it("runs a filter with predicate returning false for all", async function () {
+      await runner.run((root) => {
+        root
+          .apply(beam.create([1, 3, 5, 7]))
+          .filter((x) => x % 2 === 0)
+          .apply(testing.assertDeepEqual([]));
+      });
+    });
+
+    it("runs a filter with predicate returning true for all", async function () {
+      await runner.run((root) => {
+        root
+          .apply(beam.create([2, 4, 6, 8]))
+          .filter((x) => x % 2 === 0)
+          .apply(testing.assertDeepEqual([2, 4, 6, 8]));
+      });
+    });
+
+    it("runs a filter with context", async function () {
+      await runner.run((root) => {
+        root
+          .apply(beam.create([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))
+          .filter((x: number, threshold: number) => x > threshold, 5)
+          .apply(testing.assertDeepEqual([6, 7, 8, 9, 10]));
+      });
+    });
+
+    it("runs a filter on strings", async function () {
+      await runner.run((root) => {
+        root
+          .apply(beam.create(["apple", "banana", "apricot", "cherry"]))
+          .filter((s) => s.startsWith("a"))
+          .apply(testing.assertDeepEqual(["apple", "apricot"]));
+      });
+    });
+
     it("runs a Splitter", async function () {
       await runner.run((root) => {
         const pcolls = root