Merge pull request #33948 SnowflakeIO: be consistent with backslash escape char.

Author: robertwb

sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/SnowflakeIO.java

@@ -1176,7 +1176,10 @@ public void processElement(ProcessContext context) {
       for (Object o : context.element()) {
         if (o instanceof String) {
           String field = (String) o;
-          field = field.replace("'", "''");
+          field = field.replace("\\", "\\\\");
+          if (!this.quotationMark.isEmpty()) {
+            field = field.replace(this.quotationMark, "\\" + this.quotationMark);
+          }
           field = quoteNonEmptyField(field);
 
           csvItems.add(field);

sdks/java/io/snowflake/src/test/java/org/apache/beam/sdk/io/snowflake/test/unit/write/SnowflakeIOWriteTest.java

@@ -208,7 +208,7 @@ public void writeToExternalWithDoubleQuotation() throws SnowflakeSQLException {
     List<String> actualData = FakeSnowflakeDatabase.getElements(FAKE_TABLE);
     List<String> escapedTestData =
         testDataInStrings.stream()
-            .map(e -> e.replace("'", "''"))
+            .map(e -> e.replace("\"", "\\\""))
             .map(e -> e.isEmpty() ? "" : String.format("\"%s\"", e))
             .collect(Collectors.toList());
     assertTrue(TestUtils.areListsEqual(escapedTestData, actualData));
@@ -233,8 +233,7 @@ public void writeToExternalWithBlankQuotation() throws SnowflakeSQLException {
 
     List<String> actualData = FakeSnowflakeDatabase.getElements(FAKE_TABLE);
 
-    List<String> escapedTestData =
-        testDataInStrings.stream().map(e -> e.replace("'", "''")).collect(Collectors.toList());
-    assertTrue(TestUtils.areListsEqual(escapedTestData, actualData));
+    // no escape for blank quotation
+    assertTrue(TestUtils.areListsEqual(testDataInStrings, actualData));
   }
 }