[Data Validation] Add table configuration

v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/config/TableConfiguration.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright (C) 2026 Google LLC
+ *
+ * Licensed 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 com.google.cloud.teleport.v2.config;
+
+import com.google.cloud.teleport.v2.options.GCSSpannerDVOptions;
+import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaMapper;
+import com.google.gson.Gson;
+import java.io.InputStream;
+import java.io.Serializable;
+import java.nio.channels.Channels;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.ResourceId;
+import org.apache.commons.io.IOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Configuration class for table-based filtering in Data Validation pipeline. Encapsulates parsing,
+ * matching, and validation of source and Spanner tables.
+ */
+public class TableConfiguration implements Serializable {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TableConfiguration.class);
+
+  private final Set<String> configuredSourceTables;
+
+  private TableConfiguration(Set<String> configuredSourceTables) {
+    this.configuredSourceTables = Collections.unmodifiableSet(configuredSourceTables);
+  }
+
+  /** Creates an empty configuration with no filters. Useful for testing. */
+  public static TableConfiguration empty() {
+    return new TableConfiguration(new HashSet<>());
+  }
+
+  /**
+   * Parses and validates table configuration from pipeline options.
+   *
+   * @param options The pipeline options.
+   * @return A TableConfiguration instance containing the configured source tables.
+   */
+  public static TableConfiguration parseFromOptions(GCSSpannerDVOptions options) {
+    String tablesConfig = options.getTables();
+    String tableConfigurationFilePath = options.getTableConfigurationFilePath();
+    boolean hasTablesConfig = tablesConfig != null && !tablesConfig.trim().isEmpty();
+    boolean hasTableConfigFile =
+        tableConfigurationFilePath != null && !tableConfigurationFilePath.trim().isEmpty();
+
+    if (hasTablesConfig && hasTableConfigFile) {
+      throw new IllegalArgumentException(
+          "Both --tables and --tableConfigurationFilePath are provided. Please configure only one of these parameters at a time.");
+    }
+
+    Set<String> configuredTables = new HashSet<>();
+
+    if (hasTablesConfig) {
+      for (String table : tablesConfig.split(",")) {
+        String trimmed = table.trim();
+        if (!trimmed.isEmpty()) {
+          configuredTables.add(trimmed);
+        }
+      }
+    } else if (hasTableConfigFile) {
+      try {
+        ResourceId resourceId = FileSystems.matchNewResource(tableConfigurationFilePath, false);
+        try (InputStream stream = Channels.newInputStream(FileSystems.open(resourceId))) {
+          String result = IOUtils.toString(stream, StandardCharsets.UTF_8);
+          Gson gson = new Gson();
+          TableConfigurationFile fileConfig = gson.fromJson(result, TableConfigurationFile.class);
+
+          if (fileConfig != null && fileConfig.getTableNames() != null) {
+            for (String table : fileConfig.getTableNames()) {
+              String trimmed = table.trim();
+              if (!trimmed.isEmpty()) {
+                configuredTables.add(trimmed);
+              }
+            }
+          }
+        }
+      } catch (Exception e) {
+        throw new RuntimeException(
+            "Failed to read JSON tableConfigurationFilePath: " + tableConfigurationFilePath, e);
+      }
+    }
+
+    TableConfiguration config = new TableConfiguration(configuredTables);
+
+    return config;
+  }
+
+  public boolean hasFilters() {
+    return configuredSourceTables != null && !configuredSourceTables.isEmpty();
+  }
+
+  public Set<String> getSourceTables() {
+    return configuredSourceTables;
+  }
+
+  /**
+   * Checks if a source table is allowed by the configuration.
+   *
+   * @param sourceTableName The source table name.
+   * @return true if allowed or no filters are configured, false otherwise.
+   */
+  public boolean isSourceTableAllowed(String sourceTableName) {
+    if (!hasFilters()) {
+      return true;
+    }
+    return configuredSourceTables.contains(sourceTableName);
+  }
+
+  /**
+   * Checks if a Spanner table is allowed by the configuration. Translates the Spanner table name to
+   * its source table counterpart using the schema mapper.
+   *
+   * @param spannerTableName The Spanner table name.
+   * @param schemaMapper The schema mapper to translate the table name.
+   * @return true if allowed or no filters are configured, false otherwise.
+   */
+  public boolean isSpannerTableAllowed(String spannerTableName, ISchemaMapper schemaMapper) {
+    if (!hasFilters()) {
+      return true;
+    }
+    try {
+      String sourceTable = schemaMapper.getSourceTableName("", spannerTableName);
+      return configuredSourceTables.contains(sourceTable);
+    } catch (NoSuchElementException e) {
+      LOG.warn(
+          "Could not map Spanner table '{}' back to a source table. Skipping validation.",
+          spannerTableName);
+      return false;
+    }
+  }
+}

v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/config/TableConfigurationFile.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright (C) 2026 Google LLC
+ *
+ * Licensed 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 com.google.cloud.teleport.v2.config;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/** POJO representing the table configuration JSON file. */
+public class TableConfigurationFile implements Serializable {
+
+  private final List<String> tableNames;
+
+  /**
+   * Future Extensibility: Map of Source Table Name -> Table-specific configuration.
+   *
+   * <p>Note: The `tableNames` list remains the absolute source of truth for the exhaustive list of
+   * tables to be validated. This map is strictly for providing advanced configurations (e.g.,
+   * column filtering, sampling) for a subset of those tables. Tables cannot be implicitly included
+   * for validation by solely appearing in this map; they MUST be explicitly listed in `tableNames`.
+   *
+   * <p>This is currently a placeholder and is not yet processed by the pipeline logic.
+   */
+  private final Map<String, TableLevelConfig> optionalConfigurations;
+
+  public TableConfigurationFile(
+      List<String> tableNames, Map<String, TableLevelConfig> optionalConfigurations) {
+    this.tableNames = tableNames;
+    this.optionalConfigurations = optionalConfigurations;
+  }
+
+  public List<String> getTableNames() {
+    return tableNames;
+  }
+
+  public Map<String, TableLevelConfig> getOptionalConfigurations() {
+    return optionalConfigurations;
+  }
+}

v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/config/TableLevelConfig.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) 2026 Google LLC
+ *
+ * Licensed 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 com.google.cloud.teleport.v2.config;
+
+import java.io.Serializable;
+
+/**
+ * Placeholder POJO representing future advanced configurations for a specific table.
+ *
+ * <p>This is intended to support features like column-level validation or deterministic sampling in
+ * the future.
+ */
+public class TableLevelConfig implements Serializable {
+
+  // Intentionally left empty for now.
+  //
+  // Example future fields:
+  // private List<String> columnsToValidate;
+  // private SamplingConfig sampling;
+
+}

v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/config/package-info.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright (C) 2026 Google LLC
+ *
+ * Licensed 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.
+ */
+
+/** Configuration classes for Data Validation pipeline. */
+package com.google.cloud.teleport.v2.config;

v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/dofn/CreateSpannerReadOpsFn.java

@@ -15,35 +15,48 @@
  */
 package com.google.cloud.teleport.v2.dofn;
 
+import com.google.cloud.teleport.v2.config.TableConfiguration;
 import com.google.cloud.teleport.v2.spanner.ddl.Ddl;
+import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaMapper;
 import java.util.List;
 import org.apache.beam.sdk.io.gcp.spanner.ReadOperation;
 import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.values.PCollectionView;
 
 public class CreateSpannerReadOpsFn extends DoFn<Void, ReadOperation> {
 
   private final PCollectionView<Ddl> ddlView;
+  private final SerializableFunction<Ddl, ISchemaMapper> schemaMapperProvider;
+  private final TableConfiguration tableConfig;
 
-  public CreateSpannerReadOpsFn(PCollectionView<Ddl> ddlView) {
+  public CreateSpannerReadOpsFn(
+      PCollectionView<Ddl> ddlView,
+      SerializableFunction<Ddl, ISchemaMapper> schemaMapperProvider,
+      TableConfiguration tableConfig) {
     this.ddlView = ddlView;
+    this.schemaMapperProvider = schemaMapperProvider;
+    this.tableConfig = tableConfig;
   }
 
   // TODO: @aasthabharill to check if there's a better way to generalize dialect specific changes
   @ProcessElement
   public void processElement(ProcessContext c) {
     Ddl ddl = c.sideInput(ddlView);
+    ISchemaMapper schemaMapper = schemaMapperProvider.apply(ddl);
     List<String> tableNames = ddl.getTablesOrderedByReference();
-    tableNames.forEach(
-        tableName -> {
-          String quote = ddl.dialect() == com.google.cloud.spanner.Dialect.POSTGRESQL ? "\"" : "`";
-          // We encode the tableName in the query itself to push table information dynamically
-          // and avoid table level stages.
-          String query =
-              String.format(
-                  "SELECT *, '%s' as __tableName__ FROM %s%s%s",
-                  tableName, quote, tableName, quote);
-          c.output(ReadOperation.create().withQuery(query));
-        });
+
+    for (String tableName : tableNames) {
+      if (tableConfig != null && !tableConfig.isSpannerTableAllowed(tableName, schemaMapper)) {
+        continue;
+      }
+      String quote = ddl.dialect() == com.google.cloud.spanner.Dialect.POSTGRESQL ? "\"" : "`";
+      // We encode the tableName in the query itself to push table information dynamically
+      // and avoid table level stages.
+      String query =
+          String.format(
+              "SELECT *, '%s' as __tableName__ FROM %s%s%s", tableName, quote, tableName, quote);
+      c.output(ReadOperation.create().withQuery(query));
+    }
   }
 }

v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/options/GCSSpannerDVOptions.java

@@ -215,4 +215,25 @@ public interface GCSSpannerDVOptions extends PipelineOptions {
   String getTransformationCustomParameters();
 
   void setTransformationCustomParameters(String value);
+
+  @TemplateParameter.Text(
+      order = 16,
+      optional = true,
+      description = "Comma-separated list of source tables to validate",
+      helpText = "A comma-separated list of source tables to include in the validation run.")
+  @Default.String("")
+  String getTables();
+
+  void setTables(String value);
+
+  @TemplateParameter.GcsReadFile(
+      order = 17,
+      optional = true,
+      description = "GCS path to a file containing a list of source tables to validate",
+      helpText =
+          "A GCS file path containing a JSON list of source tables to validate. This must be a JSON file with the structure `{\"tableNames\": [\"table1\", \"table2\"]}`.")
+  @Default.String("")
+  String getTableConfigurationFilePath();
+
+  void setTableConfigurationFilePath(String value);
 }

v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/templates/GCSSpannerDV.java

@@ -21,6 +21,7 @@
 import com.google.cloud.teleport.metadata.Template;
 import com.google.cloud.teleport.metadata.TemplateCategory;
 import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
+import com.google.cloud.teleport.v2.config.TableConfiguration;
 import com.google.cloud.teleport.v2.dto.ComparisonRecord;
 import com.google.cloud.teleport.v2.fn.SchemaMapperProviderFn;
 import com.google.cloud.teleport.v2.options.GCSSpannerDVOptions;
@@ -76,6 +77,8 @@ public static void main(String[] args) {
   public static PipelineResult run(GCSSpannerDVOptions options) {
     Pipeline pipeline = Pipeline.create(options);
 
+    TableConfiguration tableConfig = TableConfiguration.parseFromOptions(options);
+
     SpannerConfig spannerConfig = createSpannerConfig(options);
 
     // Fetch Spanner DDL using Info schema
@@ -107,13 +110,14 @@ public static PipelineResult run(GCSSpannerDVOptions options) {
                 options.getGcsInputDirectory(),
                 ddlView,
                 schemaMapperProvider,
-                customTransformation));
+                customTransformation,
+                tableConfig));
 
     // Get Spanner records hashes
     PCollection<ComparisonRecord> spannerRecords =
         pipeline.apply(
             "ReadSpannerRecords",
-            new SpannerReaderTransform(spannerConfig, ddlView, schemaMapperProvider));
+            new SpannerReaderTransform(spannerConfig, ddlView, schemaMapperProvider, tableConfig));
 
     PCollectionTuple inputs =
         PCollectionTuple.of(SOURCE_TAG, sourceRecords).and(SPANNER_TAG, spannerRecords);