Add filter parameter to Iceberg read configuration- Added filter (str) parameter to ReadFromIceberg transform in IcebergToPostgres.yaml- Updated IcebergToPostgresYaml.java interface with filter parameter between drop and keep- Adjusted order numbers for subsequent parameters- Added filter parameter to iceberg_options.yaml with example

Author: Ulada Butsenka

yaml/src/main/java/com/google/cloud/teleport/templates/yaml/IcebergToPostgresYaml.java

@@ -99,6 +99,16 @@ public interface IcebergToPostgresYaml {
 
   @TemplateParameter.Text(
       order = 6,
+      name = "filter",
+      optional = true,
+      description =
+          "An optional filter expression to apply to the input records from the Iceberg table.",
+      helpText = "A filter expression to apply to records from the Iceberg table.",
+      example = "age > 18")
+  String getFilter();
+
+  @TemplateParameter.Text(
+      order = 7,
       name = "keep",
       optional = true,
       description =
@@ -109,7 +119,7 @@ public interface IcebergToPostgresYaml {
   String getKeep();
 
   @TemplateParameter.Text(
-      order = 7,
+      order = 8,
       name = "jdbcUrl",
       optional = false,
       description = "Connection URL for the JDBC source/sink.",
@@ -119,7 +129,7 @@ public interface IcebergToPostgresYaml {
   String getJdbcUrl();
 
   @TemplateParameter.Text(
-      order = 8,
+      order = 9,
       name = "username",
       optional = true,
       description = "Username for the JDBC connection.",
@@ -128,7 +138,7 @@ public interface IcebergToPostgresYaml {
   String getUsername();
 
   @TemplateParameter.Password(
-      order = 9,
+      order = 10,
       name = "password",
       optional = true,
       description = "Password for the JDBC connection.",
@@ -137,7 +147,7 @@ public interface IcebergToPostgresYaml {
   String getPassword();
 
   @TemplateParameter.Text(
-      order = 10,
+      order = 11,
       name = "driverClassName",
       optional = true,
       description =
@@ -148,7 +158,7 @@ public interface IcebergToPostgresYaml {
   String getDriverClassName();
 
   @TemplateParameter.Text(
-      order = 11,
+      order = 12,
       name = "driverJars",
       optional = true,
       description = "Comma-separated GCS paths of the JDBC driver jars.",
@@ -157,7 +167,7 @@ public interface IcebergToPostgresYaml {
   String getDriverJars();
 
   @TemplateParameter.Text(
-      order = 12,
+      order = 13,
       name = "connectionProperties",
       optional = true,
       description = "JDBC connection properties.",
@@ -166,7 +176,7 @@ public interface IcebergToPostgresYaml {
   String getConnectionProperties();
 
   @TemplateParameter.Text(
-      order = 13,
+      order = 14,
       name = "connectionInitSql",
       optional = true,
       description = "A list of SQL statements to execute upon connection initialization.",
@@ -175,7 +185,7 @@ public interface IcebergToPostgresYaml {
   String getConnectionInitSql();
 
   @TemplateParameter.Text(
-      order = 14,
+      order = 15,
       name = "jdbcType",
       optional = true,
       description = "Type of JDBC source. Default: postgres.",
@@ -186,7 +196,7 @@ public interface IcebergToPostgresYaml {
   String getJdbcType();
 
   @TemplateParameter.Text(
-      order = 15,
+      order = 16,
       name = "location",
       optional = false,
       description = "The name of the table to write to.",
@@ -195,7 +205,7 @@ public interface IcebergToPostgresYaml {
   String getLocation();
 
   @TemplateParameter.Text(
-      order = 16,
+      order = 17,
       name = "writeStatement",
       optional = true,
       description = "The SQL statement to use for inserting records.",
@@ -204,7 +214,7 @@ public interface IcebergToPostgresYaml {
   String getWriteStatement();
 
   @TemplateParameter.Integer(
-      order = 17,
+      order = 18,
       name = "batchSize",
       optional = true,
       description = "The number of records to group for each write operation.",
@@ -213,7 +223,7 @@ public interface IcebergToPostgresYaml {
   Integer getBatchSize();
 
   @TemplateParameter.Boolean(
-      order = 18,
+      order = 19,
       name = "autosharding",
       optional = true,
       description = "If true, enables using a dynamically determined number of shards to write.",

yaml/src/main/python/options/iceberg_options.yaml

@@ -37,6 +37,13 @@ options:
         required: false
         type: text
       - order: 6
+        name: "filter"
+        description: "An optional filter expression to apply to the input records."
+        help: "A filter expression to apply to records from the Iceberg table."
+        example: "age > 18"
+        required: false
+        type: text
+      - order: 7
         name: "keep"
         description: "A list of field names to keep in the input record."
         help: "A list of field names to keep. Mutually exclusive with 'drop' and 'only'."

yaml/src/main/yaml/IcebergToPostgres.yaml

@@ -40,6 +40,7 @@ pipeline:
         catalog_properties: {{ catalogProperties }}
         config_properties: {{ configProperties }}
         drop: {{ drop }}
+        filter: {{ filter }}
         keep: {{ keep }}
 
     - type: WriteToPostgres

yaml/src/test/java/com/google/cloud/teleport/templates/yaml/IcebergToPostgresYamlIT.java

@@ -56,15 +56,16 @@ public class IcebergToPostgresYamlIT extends TemplateTestBase {
   private static final Logger LOG = LoggerFactory.getLogger(IcebergToPostgresYamlIT.class);
 
   // Iceberg Setup
-  private static final String CATALOG_NAME = "hadoop_catalog";
+  private static final String CATALOG_NAME = "gcs_catalog";
   private static final String NAMESPACE = "iceberg_namespace";
   private static final String ICEBERG_TABLE_NAME = "source_table";
   private static final String ICEBERG_TABLE_IDENTIFIER = NAMESPACE + "." + ICEBERG_TABLE_NAME;
 
   // Postgres Setup
   private static final String POSTGRES_TABLE_NAME = "target_table";
-  private static final String WRITE_STATEMENT =
-      "INSERT INTO " + POSTGRES_TABLE_NAME + " (id, name, active) VALUES (?, ?, ?)";
+
+  //   private static final String WRITE_STATEMENT =
+  //       "INSERT INTO " + POSTGRES_TABLE_NAME + " (id, name, active) VALUES (?, ?, ?)";
 
   private PostgresResourceManager postgresResourceManager;
   private IcebergResourceManager icebergResourceManager;
@@ -135,6 +136,7 @@ public void testIcebergToPostgres() throws IOException {
             .addParameter("jdbcUrl", postgresResourceManager.getUri())
             .addParameter("username", postgresResourceManager.getUsername())
             .addParameter("password", postgresResourceManager.getPassword())
+            // .addParameter("writeStatement", WRITE_STATEMENT)
             .addParameter("location", POSTGRES_TABLE_NAME);
 
     // Act
@@ -188,13 +190,6 @@ protected PipelineOperator.Config createConfig(LaunchInfo info) {
         .build();
   }
 
-  /**
-   * Build catalog properties for Iceberg warehouse backed by GCS.
-   *
-   * <p>Uses BigLake REST catalog for Iceberg metadata and GCS for warehouse storage.
-   *
-   * @return Map of catalog properties required by Iceberg
-   */
   private Map<String, String> getCatalogProperties() {
     return Map.of(
         "type", "rest",