Merge remote-tracking branch 'upstream/main' into cassandra_rr_terraf…

…orm_sample

.github/actions/publish-site-report/action.yml

@@ -32,7 +32,7 @@ runs:
       shell: bash
       run: mvn -B surefire-report:report-only -f pom.xml -Daggregate=true -Denforcer.skip=true
     - name: Publish Site Report
-      uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+      uses: actions/upload-artifact@v4
       with:
         name: ${{ inputs.output-zip-file }}
         path: 'target/site/'

.github/workflows/java-pr.yml

@@ -104,10 +104,10 @@ jobs:
       - name: Run Unit Tests
         run: ./cicd/run-unit-tests
       - name: Upload Unit Tests Report
-        uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+        uses: actions/upload-artifact@v4
         if: always() # always run even if the previous step fails
         with:
-          name: surefire-test-results
+          name: surefire-unit-test-results
           path: '**/surefire-reports/TEST-*.xml'
           retention-days: 1
       - name: Upload coverage reports to Codecov
@@ -142,10 +142,10 @@ jobs:
           --it-artifact-bucket="cloud-teleport-testing-it-gitactions" \
           --it-private-connectivity="datastream-private-connect-us-central1"
       - name: Upload Smoke Tests Report
-        uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+        uses: actions/upload-artifact@v4
         if: always() # always run even if the previous step fails
         with:
-          name: surefire-test-results
+          name: surefire-smoke-test-results
           path: '**/surefire-reports/TEST-*.xml'
           retention-days: 1
       - name: Cleanup Java Environment
@@ -172,10 +172,10 @@ jobs:
           --it-artifact-bucket="cloud-teleport-testing-it-gitactions" \
           --it-private-connectivity="datastream-private-connect-us-central1"
       - name: Upload Integration Tests Report
-        uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+        uses: actions/upload-artifact@v4
         if: always() # always run even if the previous step fails
         with:
-          name: surefire-test-results
+          name: surefire-integration-test-results
           path: '**/surefire-reports/TEST-*.xml'
           retention-days: 1
       - name: Cleanup Java Environment

.github/workflows/kafka-pr.yml

@@ -107,10 +107,10 @@ jobs:
           ./cicd/run-unit-tests \
           --modules-to-build="KAFKA"
       - name: Upload Unit Tests Report
-        uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+        uses: actions/upload-artifact@v4
         if: always() # always run even if the previous step fails
         with:
-          name: surefire-test-results
+          name: surefire-unit-test-results
           path: '**/surefire-reports/TEST-*.xml'
           retention-days: 1
       - name: Upload coverage reports to Codecov
@@ -144,10 +144,10 @@ jobs:
           --it-artifact-bucket="cloud-teleport-testing-it-gitactions" \
           --it-private-connectivity="datastream-private-connect-us-central1"
       - name: Upload Smoke Tests Report
-        uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+        uses: actions/upload-artifact@v4
         if: always() # always run even if the previous step fails
         with:
-          name: surefire-test-results
+          name: surefire-smoke-test-results
           path: '**/surefire-reports/TEST-*.xml'
           retention-days: 1
       - name: Cleanup Java Environment
@@ -173,10 +173,10 @@ jobs:
           --it-artifact-bucket="cloud-teleport-testing-it-gitactions" \
           --it-private-connectivity="datastream-private-connect-us-central1"
       - name: Upload Integration Tests Report
-        uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+        uses: actions/upload-artifact@v4
         if: always() # always run even if the previous step fails
         with:
-          name: surefire-test-results
+          name: surefire-integration-test-results
           path: '**/surefire-reports/TEST-*.xml'
           retention-days: 1
       - name: Cleanup Java Environment

.github/workflows/load-tests.yml

@@ -38,7 +38,7 @@ jobs:
       env:
         HOST_IP: ${{ steps.variables.outputs.hostIP }}
     - name: Upload Load Tests Report
-      uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+      uses: actions/upload-artifact@v4
       if: always() # always run even if the previous step fails
       with:
         name: surefire-test-results

.github/workflows/scorecards-analysis.yml

@@ -42,7 +42,7 @@ jobs:
 
       # Upload the results as artifacts (optional).
       - name: "Upload artifact"
-        uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+        uses: actions/upload-artifact@v4
         with:
           name: SARIF file
           path: results.sarif

.github/workflows/spanner-load-tests.yml

@@ -62,10 +62,10 @@ jobs:
       with:
         filename: .github/ISSUE_TEMPLATE/spanner-load-test-failure-issue-template.md
     - name: Upload Load Tests Report
-      uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+      uses: actions/upload-artifact@v4
       if: always() # always run even if the previous step fails
       with:
-        name: surefire-test-results
+        name: surefire-load-test-results
         path: '**/surefire-reports/TEST-*.xml'
         retention-days: 30
     - name: Load Test report on GitHub
@@ -117,11 +117,11 @@ jobs:
       with:
         filename: .github/ISSUE_TEMPLATE/spanner-load-test-failure-issue-template.md
     - name: Upload Load Test Observer Report
-      uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+      uses: actions/upload-artifact@v4
       if: always() # always run even if the previous step fails
       with:
-        name: surefire-test-results
+        name: surefire-observer-test-results
         path: '**/surefire-reports/TEST-*.xml'
         retention-days: 1
     - name: Cleanup Java Environment
-      uses: ./.github/actions/cleanup-java-env
+      uses: ./.github/actions/cleanup-java-env

.github/workflows/spanner-pr.yml

@@ -111,10 +111,10 @@ jobs:
         ./cicd/run-unit-tests \
         --modules-to-build="SPANNER"
     - name: Upload Unit Tests Report
-      uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+      uses: actions/upload-artifact@v4
       if: always() # always run even if the previous step fails
       with:
-        name: surefire-test-results
+        name: surefire-unit-test-results
         path: '**/surefire-reports/TEST-*.xml'
         retention-days: 1
     - name: Upload coverage reports to Codecov
@@ -148,10 +148,10 @@ jobs:
         --it-artifact-bucket="cloud-teleport-testing-it-gitactions" \
         --it-private-connectivity="datastream-private-connect-us-central1"
     - name: Upload Smoke Tests Report
-      uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+      uses: actions/upload-artifact@v4
       if: always() # always run even if the previous step fails
       with:
-        name: surefire-test-results
+        name: surefire-smoke-test-results
         path: '**/surefire-reports/TEST-*.xml'
         retention-days: 1
     - name: Cleanup Java Environment
@@ -177,10 +177,10 @@ jobs:
         --it-artifact-bucket="cloud-teleport-testing-it-gitactions" \
         --it-private-connectivity="datastream-private-connect-us-central1"
     - name: Upload Integration Tests Report
-      uses: actions/upload-artifact@0b7f8abb1508181956e8e162db84b466c27e18ce # v3.1.2
+      uses: actions/upload-artifact@v4
       if: always() # always run even if the previous step fails
       with:
-        name: surefire-test-results
+        name: surefire-integration-test-results
         path: '**/surefire-reports/TEST-*.xml'
         retention-days: 10
     - name: Cleanup Java Environment

v1/src/main/java/com/google/cloud/teleport/spanner/ddl/Index.java

@@ -85,7 +85,9 @@ public void prettyPrint(Appendable appendable) throws IOException {
 
   private void prettyPrintPg(Appendable appendable) throws IOException {
     appendable.append("CREATE");
-    if (unique()) {
+    if (type() != null && (type().equals("SEARCH"))) {
+      appendable.append(" " + type());
+    } else if (unique()) {
       appendable.append(" UNIQUE");
     }
     appendable
@@ -111,13 +113,42 @@ private void prettyPrintPg(Appendable appendable) throws IOException {
       appendable.append(" INCLUDE (").append(storingString).append(")");
     }
 
+    if (partitionBy() != null) {
+      String partitionByString =
+          partitionBy().stream()
+              .map(c -> quoteIdentifier(c, dialect()))
+              .collect(Collectors.joining(","));
+
+      if (!partitionByString.isEmpty()) {
+        appendable.append(" PARTITION BY ").append(partitionByString);
+      }
+    }
+
+    if (orderBy() != null) {
+      String orderByString =
+          orderBy().stream()
+              .map(c -> quoteIdentifier(c, dialect()))
+              .collect(Collectors.joining(","));
+
+      if (!orderByString.isEmpty()) {
+        appendable.append(" ORDER BY ").append(orderByString);
+      }
+    }
+
     if (interleaveIn() != null) {
       appendable.append(" INTERLEAVE IN ").append(quoteIdentifier(interleaveIn(), dialect()));
     }
 
     if (filter() != null && !filter().isEmpty()) {
       appendable.append(" WHERE ").append(filter());
     }
+
+    if (options() != null) {
+      String optionsString = String.join(",", options());
+      if (!optionsString.isEmpty()) {
+        appendable.append(" WITH (").append(optionsString).append(")");
+      }
+    }
   }
 
   private void prettyPrintGsql(Appendable appendable) throws IOException {

v1/src/main/java/com/google/cloud/teleport/spanner/ddl/InformationSchemaScanner.java

@@ -118,9 +118,7 @@ public Ddl scan() {
     Map<String, NavigableMap<String, Index.Builder>> indexes = Maps.newHashMap();
     listIndexes(indexes);
     listIndexColumns(builder, indexes);
-    if (dialect == Dialect.GOOGLE_STANDARD_SQL) {
-      listIndexOptions(builder, indexes);
-    }
+    listIndexOptions(builder, indexes);
 
     for (Map.Entry<String, NavigableMap<String, Index.Builder>> tableEntry : indexes.entrySet()) {
       String tableName = tableEntry.getKey();
@@ -364,10 +362,7 @@ private void listColumns(Ddl.Builder builder) {
           dialect == Dialect.GOOGLE_STANDARD_SQL
               ? resultSet.getBoolean(15)
               : resultSet.getString(15).equalsIgnoreCase("YES");
-      boolean isPlacementKey =
-          dialect == Dialect.GOOGLE_STANDARD_SQL
-              ? resultSet.getBoolean(16)
-              : resultSet.getBoolean(16);
+      boolean isPlacementKey = resultSet.getBoolean(16);
 
       builder
           .createTable(tableName)
@@ -463,20 +458,15 @@ private void listIndexes(Map<String, NavigableMap<String, Index.Builder>> indexe
               : resultSet.getString(5).equalsIgnoreCase("YES");
       String filter = resultSet.isNull(6) ? null : resultSet.getString(6);
 
-      // Note that 'type' is only queried from GoogleSQL and is not from Postgres and
-      // the number of columns will be different.
-      String type =
-          (dialect == Dialect.GOOGLE_STANDARD_SQL && !resultSet.isNull(7))
-              ? resultSet.getString(7)
-              : null;
+      String type = !resultSet.isNull(7) ? resultSet.getString(7) : null;
 
       ImmutableList<String> searchPartitionBy =
-          (dialect == Dialect.GOOGLE_STANDARD_SQL && !resultSet.isNull(8))
+          !resultSet.isNull(8)
               ? ImmutableList.<String>builder().addAll(resultSet.getStringList(8)).build()
               : null;
 
       ImmutableList<String> searchOrderBy =
-          (dialect == Dialect.GOOGLE_STANDARD_SQL && !resultSet.isNull(9))
+          !resultSet.isNull(9)
               ? ImmutableList.<String>builder().addAll(resultSet.getStringList(9)).build()
               : null;
 
@@ -513,10 +503,11 @@ Statement listIndexesSQL() {
       case POSTGRESQL:
         return Statement.of(
             "SELECT t.table_schema, t.table_name, t.index_name, t.parent_table_name, t.is_unique,"
-                + " t.is_null_filtered, t.filter FROM information_schema.indexes AS t "
+                + " t.is_null_filtered, t.filter, t.index_type, t.search_partition_by, t.search_order_by"
+                + " FROM information_schema.indexes AS t "
                 + " WHERE t.table_schema NOT IN "
                 + " ('information_schema', 'spanner_sys', 'pg_catalog')"
-                + " AND t.index_type='INDEX' AND t.spanner_is_managed = 'NO' "
+                + " AND (t.index_type='INDEX' OR t.index_type='SEARCH') AND t.spanner_is_managed = 'NO' "
                 + " ORDER BY t.table_name, t.index_name");
       default:
         throw new IllegalArgumentException("Unrecognized dialect: " + dialect);
@@ -533,8 +524,8 @@ private void listIndexColumns(
       String columnName = resultSet.getString(2);
       String ordering = resultSet.isNull(3) ? null : resultSet.getString(3);
       String indexLocalName = resultSet.getString(4);
-      String indexType = dialect == Dialect.GOOGLE_STANDARD_SQL ? resultSet.getString(5) : null;
-      String spannerType = dialect == Dialect.GOOGLE_STANDARD_SQL ? resultSet.getString(6) : null;
+      String indexType = resultSet.getString(5);
+      String spannerType = resultSet.getString(6);
 
       if (indexLocalName.equals("PRIMARY_KEY")) {
         IndexColumn.IndexColumnsBuilder<Table.Builder> pkBuilder =
@@ -546,8 +537,10 @@ private void listIndexColumns(
         }
         pkBuilder.end().endTable();
       } else {
+        String tokenlistType = dialect == Dialect.POSTGRESQL ? "spanner.tokenlist" : "TOKENLIST";
         if (indexType != null && ordering != null) {
-          if ((indexType.equals("SEARCH") && !spannerType.equals("TOKENLIST"))
+          // Non-tokenlist columns should not be included in the key for Search Indexes.
+          if ((indexType.equals("SEARCH") && !spannerType.contains(tokenlistType))
               || (indexType.equals("VECTOR") && !spannerType.startsWith("ARRAY"))) {
             continue;
           }
@@ -567,8 +560,9 @@ private void listIndexColumns(
         }
         IndexColumn.IndexColumnsBuilder<Index.Builder> indexColumnsBuilder =
             indexBuilder.columns().create().name(columnName);
+        // Tokenlist columns do not have ordering.
         if (spannerType != null
-            && (spannerType.equals("TOKENLIST") || spannerType.startsWith("ARRAY"))) {
+            && (spannerType.equals(tokenlistType) || spannerType.startsWith("ARRAY"))) {
           indexColumnsBuilder.none();
         } else if (ordering == null) {
           indexColumnsBuilder.storing();
@@ -605,7 +599,8 @@ Statement listIndexColumnsSQL() {
                 + "ORDER BY t.table_name, t.index_name, t.ordinal_position");
       case POSTGRESQL:
         return Statement.of(
-            "SELECT t.table_schema, t.table_name, t.column_name, t.column_ordering, t.index_name "
+            "SELECT t.table_schema, t.table_name, t.column_name, t.column_ordering, t.index_name,"
+                + " t.index_type, t.spanner_type "
                 + "FROM information_schema.index_columns AS t "
                 + "WHERE t.table_schema NOT IN "
                 + "('information_schema', 'spanner_sys', 'pg_catalog') "
@@ -674,6 +669,14 @@ Statement listIndexOptionsSQL() {
                 + " WHERE t.table_schema NOT IN"
                 + " ('INFORMATION_SCHEMA', 'SPANNER_SYS')"
                 + " ORDER BY t.table_name, t.index_name, t.option_name");
+      case POSTGRESQL:
+        return Statement.of(
+            "SELECT t.table_schema, t.table_name, t.index_name, t.index_type,"
+                + " t.option_name, t.option_type, t.option_value"
+                + " FROM information_schema.index_options AS t"
+                + " WHERE t.table_schema NOT IN"
+                + " ('information_schema', 'spanner_sys', 'pg_catalog') "
+                + " ORDER BY t.table_name, t.index_name, t.option_name");
       default:
         throw new IllegalArgumentException("Unrecognized dialect: " + dialect);
     }

v1/src/test/java/com/google/cloud/teleport/spanner/DdlToAvroSchemaConverterTest.java

@@ -388,7 +388,10 @@ public void pgSimple() {
             .asc("last_name")
             .end()
             .indexes(
-                ImmutableList.of("CREATE INDEX \"UsersByFirstName\" ON \"Users\" (\"first_name\")"))
+                ImmutableList.of(
+                    "CREATE INDEX \"UsersByFirstName\" ON \"Users\" (\"first_name\")",
+                    "CREATE SEARCH INDEX \"SearchIndex\" ON \"Users\" (\"tokens\")"
+                        + " WITH (sort_order_sharding=TRUE)"))
             .foreignKeys(
                 ImmutableList.of(
                     "ALTER TABLE \"Users\" ADD CONSTRAINT \"fk\" FOREIGN KEY (\"first_name\")"
@@ -496,6 +499,12 @@ public void pgSimple() {
     assertThat(
         avroSchema.getProp(SPANNER_INDEX + "0"),
         equalTo("CREATE INDEX \"UsersByFirstName\" ON \"Users\" (\"first_name\")"));
+
+    assertThat(
+        avroSchema.getProp(SPANNER_INDEX + "1"),
+        equalTo(
+            "CREATE SEARCH INDEX \"SearchIndex\" ON \"Users\" (\"tokens\") WITH (sort_order_sharding=TRUE)"));
+
     assertThat(
         avroSchema.getProp(SPANNER_FOREIGN_KEY + "0"),
         equalTo(

v1/src/test/java/com/google/cloud/teleport/spanner/ExportPipelineIT.java

@@ -278,9 +278,16 @@ private void testPostgresSpannerToGCSAvroBase(
                 + "  \"NameTokens\" spanner.tokenlist generated always as (spanner.tokenize_fulltext(\"FirstName\")) stored hidden,\n"
                 + "PRIMARY KEY(\"Id\"))",
             testName);
+    String createSearchIndexStatement =
+        String.format(
+            "CREATE SEARCH INDEX \"%s_SearchIndex\"\n"
+                + " ON \"%s_Singers\"(\"NameTokens\") ORDER BY \"Id\" WHERE \"Id\" IS NOT NULL\n"
+                + " WITH (sort_order_sharding=TRUE, disable_automatic_uid_column=TRUE)",
+            testName, testName);
 
     spannerResourceManager.executeDdlStatement(createEmptyTableStatement);
     spannerResourceManager.executeDdlStatement(createSingersTableStatement);
+    spannerResourceManager.executeDdlStatement(createSearchIndexStatement);
     List<Mutation> expectedData = generateTableRows(String.format("%s_Singers", testName));
     spannerResourceManager.write(expectedData);
     PipelineLauncher.LaunchConfig.Builder options =
@@ -305,6 +312,10 @@ private void testPostgresSpannerToGCSAvroBase(
     List<Artifact> emptyArtifacts =
         gcsClient.listArtifacts(
             "output/", Pattern.compile(String.format(".*/%s_%s.*\\.avro.*", testName, "Empty")));
+    List<Artifact> searchIndexArtifacts =
+        gcsClient.listArtifacts(
+            "output/",
+            Pattern.compile(String.format(".*/%s_%s.*\\.avro.*", testName, "SearchIndex")));
     assertThat(singersArtifacts).isNotEmpty();
     assertThat(emptyArtifacts).isNotEmpty();