Remove fastjson library

CHANGES.md

@@ -91,6 +91,7 @@
 
 * Python SDK: Legacy runner support removed from Dataflow, all pipelines must use runner v2.
 * [Python] Dataflow Runner will no longer stage Beam SDK from PyPI in the `--staging_location` at pipeline submission. Custom container images that are not based on Beam's default image must include Apache Beam installation.([#26996](https://github.com/apache/beam/issues/26996))
+* SDK Java Extensions SQL: remove fastjson library, table property is changed to ObjectNode which belongs to jackson library. ([#24154](https://github.com/apache/beam/issues/24154))
 
 ## Deprecations
 

playground/backend/containers/java/Dockerfile

@@ -65,7 +65,6 @@ ENV BEAM_SDK="SDK_JAVA"
 ENV PROPERTY_PATH=/opt/playground/backend/properties.yaml
 ARG CALCITE_VERSION=1_28_0
 ARG BYTEBUDDY_VERSION=1.12.14
-ARG FASTJSON_VERSION=1.2.69
 ARG JANINO_VERSION=3.0.11
 
 # Copy build result
@@ -103,9 +102,6 @@ RUN wget https://repo1.maven.org/maven2/org/apache/beam/beam-vendor-calcite-$CAL
 RUN wget  https://repo1.maven.org/maven2/net/bytebuddy/byte-buddy/$BYTEBUDDY_VERSION/byte-buddy-$BYTEBUDDY_VERSION.jar &&\
     mv byte-buddy-$BYTEBUDDY_VERSION.jar /opt/apache/beam/jars/byte-buddy-$BYTEBUDDY_VERSION.jar
 
-RUN wget  https://repo1.maven.org/maven2/com/alibaba/fastjson/$FASTJSON_VERSION/fastjson-$FASTJSON_VERSION.jar &&\
-    mv fastjson-$FASTJSON_VERSION.jar /opt/apache/beam/jars/fastjson-$FASTJSON_VERSION.jar
-
 RUN wget  https://repo1.maven.org/maven2/org/codehaus/janino/janino/$JANINO_VERSION/janino-$JANINO_VERSION.jar &&\
     mv janino-$JANINO_VERSION.jar /opt/apache/beam/jars/janino-$JANINO_VERSION.jar
 

sdks/java/extensions/sql/build.gradle

@@ -84,7 +84,6 @@ dependencies {
   implementation library.java.jackson_databind
   implementation library.java.joda_time
   implementation library.java.vendored_calcite_1_28_0
-  implementation "com.alibaba:fastjson:1.2.69"
   implementation "org.codehaus.janino:janino:3.0.11"
   implementation "org.codehaus.janino:commons-compiler:3.0.11"
   implementation library.java.jackson_core

sdks/java/extensions/sql/datacatalog/build.gradle

@@ -34,7 +34,7 @@ dependencies {
   implementation library.java.protobuf_java
   implementation library.java.slf4j_api
   implementation library.java.vendored_guava_32_1_2_jre
-  implementation "com.alibaba:fastjson:1.2.69"
+  implementation library.java.jackson_databind
   implementation project(path: ":sdks:java:core", configuration: "shadow")
   implementation "org.threeten:threetenbp:1.4.5"
   provided project(":sdks:java:extensions:sql")

sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/BigQueryTableFactory.java

@@ -17,14 +17,13 @@
  */
 package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
 
-import com.alibaba.fastjson.JSONObject;
 import com.google.cloud.datacatalog.v1beta1.Entry;
 import java.net.URI;
 import java.util.Optional;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.sql.TableUtils;
 import org.apache.beam.sdk.extensions.sql.meta.Table;
-import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
 
 /** {@link TableFactory} that understands Data Catalog BigQuery entries. */
 class BigQueryTableFactory implements TableFactory {
@@ -49,7 +48,7 @@ public Optional<Table.Builder> tableBuilder(Entry entry) {
     return Optional.of(
         Table.builder()
             .location(getLocation(entry))
-            .properties(new JSONObject(ImmutableMap.of("truncateTimestamps", truncateTimestamps)))
+            .properties(TableUtils.emptyProperties().put("truncateTimestamps", truncateTimestamps))
             .type("bigquery")
             .comment(""));
   }

sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/GcsTableFactory.java

@@ -17,11 +17,11 @@
  */
 package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
 
-import com.alibaba.fastjson.JSONObject;
 import com.google.cloud.datacatalog.v1beta1.Entry;
 import com.google.cloud.datacatalog.v1beta1.GcsFilesetSpec;
 import java.util.List;
 import java.util.Optional;
+import org.apache.beam.sdk.extensions.sql.TableUtils;
 import org.apache.beam.sdk.extensions.sql.meta.Table;
 
 /** {@link TableFactory} that understands Data Catalog GCS entries. */
@@ -55,7 +55,7 @@ public Optional<Table.Builder> tableBuilder(Entry entry) {
         Table.builder()
             .type("text")
             .location(filePattern)
-            .properties(new JSONObject())
+            .properties(TableUtils.emptyProperties())
             .comment(""));
   }
 }

sdks/java/extensions/sql/datacatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/datacatalog/PubsubTableFactory.java

@@ -17,12 +17,12 @@
  */
 package org.apache.beam.sdk.extensions.sql.meta.provider.datacatalog;
 
-import com.alibaba.fastjson.JSONObject;
 import com.google.cloud.datacatalog.v1beta1.Entry;
 import java.net.URI;
 import java.util.Optional;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.sql.TableUtils;
 import org.apache.beam.sdk.extensions.sql.meta.Table;
 
 /** {@link TableFactory} that understands Data Catalog Pubsub entries. */
@@ -42,7 +42,7 @@ public Optional<Table.Builder> tableBuilder(Entry entry) {
     return Optional.of(
         Table.builder()
             .location(getLocation(entry))
-            .properties(new JSONObject())
+            .properties(TableUtils.emptyProperties())
             .type("pubsub")
             .comment(""));
   }

sdks/java/extensions/sql/hcatalog/build.gradle

@@ -41,7 +41,6 @@ dependencies {
   implementation project(":sdks:java:extensions:sql")
   implementation project(":sdks:java:io:hcatalog")
   implementation project(":sdks:java:core")
-  implementation "com.alibaba:fastjson:1.2.69"
   implementation library.java.vendored_guava_32_1_2_jre
 
   testImplementation project(":sdks:java:io:hcatalog").sourceSets.test.output

sdks/java/extensions/sql/hcatalog/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/hcatalog/DatabaseProvider.java

@@ -17,7 +17,6 @@
  */
 package org.apache.beam.sdk.extensions.sql.meta.provider.hcatalog;
 
-import com.alibaba.fastjson.JSONObject;
 import java.util.Map;
 import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
 import org.apache.beam.sdk.extensions.sql.meta.Table;
@@ -80,7 +79,6 @@ public Map<String, Table> getTables() {
         .schema(tableSchema.get())
         .name(table)
         .location("")
-        .properties(new JSONObject())
         .comment("")
         .type("hcatalog")
         .build();

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/TableUtils.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.sql;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.json.JsonReadFeature;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.Map;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+
+public class TableUtils {
+  private static final ObjectMapper objectMapper =
+      JsonMapper.builder()
+          .enable(
+              JsonReadFeature.ALLOW_BACKSLASH_ESCAPING_ANY_CHARACTER,
+              JsonReadFeature.ALLOW_JAVA_COMMENTS,
+              JsonReadFeature.ALLOW_MISSING_VALUES,
+              JsonReadFeature.ALLOW_NON_NUMERIC_NUMBERS,
+              JsonReadFeature.ALLOW_LEADING_ZEROS_FOR_NUMBERS,
+              JsonReadFeature.ALLOW_SINGLE_QUOTES,
+              JsonReadFeature.ALLOW_TRAILING_COMMA,
+              JsonReadFeature.ALLOW_UNESCAPED_CONTROL_CHARS,
+              JsonReadFeature.ALLOW_UNQUOTED_FIELD_NAMES)
+          .build();
+
+  private TableUtils() {
+    // nothing here
+  }
+
+  @VisibleForTesting
+  public static ObjectMapper getObjectMapper() {
+    return objectMapper;
+  }
+
+  public static ObjectNode emptyProperties() {
+    return objectMapper.createObjectNode();
+  }
+
+  public static ObjectNode parseProperties(String json) {
+    try {
+      return (ObjectNode) objectMapper.readTree(json);
+    } catch (JsonProcessingException e) {
+      throw new RuntimeException("illegal table properties: " + json);
+    }
+  }
+
+  public static Map<String, Object> convertNode2Map(JsonNode jsonNode) {
+    return objectMapper.convertValue(jsonNode, new TypeReference<Map<String, Object>>() {});
+  }
+}

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlCreateExternalTable.java

@@ -17,13 +17,12 @@
  */
 package org.apache.beam.sdk.extensions.sql.impl.parser;
 
-import static com.alibaba.fastjson.JSON.parseObject;
 import static org.apache.beam.sdk.schemas.Schema.toSchema;
 import static org.apache.beam.vendor.calcite.v1_28_0.org.apache.calcite.util.Static.RESOURCE;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
 
-import com.alibaba.fastjson.JSONObject;
 import java.util.List;
+import org.apache.beam.sdk.extensions.sql.TableUtils;
 import org.apache.beam.sdk.extensions.sql.impl.BeamCalciteSchema;
 import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
 import org.apache.beam.sdk.extensions.sql.meta.Table;
@@ -159,8 +158,8 @@ private Table toTable() {
         .location(SqlDdlNodes.getString(location))
         .properties(
             (tblProperties == null)
-                ? new JSONObject()
-                : parseObject(SqlDdlNodes.getString(tblProperties)))
+                ? TableUtils.emptyProperties()
+                : TableUtils.parseProperties(SqlDdlNodes.getString(tblProperties)))
         .build();
   }
 }

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/Table.java

@@ -17,9 +17,10 @@
  */
 package org.apache.beam.sdk.extensions.sql.meta;
 
-import com.alibaba.fastjson.JSONObject;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.auto.value.AutoValue;
 import java.io.Serializable;
+import org.apache.beam.sdk.extensions.sql.TableUtils;
 import org.apache.beam.sdk.schemas.Schema;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -37,12 +38,12 @@ public abstract class Table implements Serializable {
 
   public abstract @Nullable String getLocation();
 
-  public abstract JSONObject getProperties();
+  public abstract ObjectNode getProperties();
 
   public abstract Builder toBuilder();
 
   public static Builder builder() {
-    return new AutoValue_Table.Builder().properties(new JSONObject());
+    return new AutoValue_Table.Builder().properties(TableUtils.emptyProperties());
   }
 
   /** Builder class for {@link Table}. */
@@ -58,7 +59,7 @@ public abstract static class Builder {
 
     public abstract Builder location(String location);
 
-    public abstract Builder properties(JSONObject properties);
+    public abstract Builder properties(ObjectNode properties);
 
     public abstract Table build();
   }

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/SchemaIOTableProviderWrapper.java

@@ -19,8 +19,8 @@
 
 import static org.apache.beam.sdk.util.RowJsonUtils.newObjectMapperWith;
 
-import com.alibaba.fastjson.JSONObject;
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.io.Serializable;
 import java.util.List;
 import org.apache.beam.sdk.annotations.Internal;
@@ -66,7 +66,7 @@ public String getTableType() {
 
   @Override
   public BeamSqlTable buildBeamSqlTable(Table tableDefinition) {
-    JSONObject tableProperties = tableDefinition.getProperties();
+    ObjectNode tableProperties = tableDefinition.getProperties();
 
     try {
       RowJson.RowJsonDeserializer deserializer =
@@ -84,8 +84,7 @@ public BeamSqlTable buildBeamSqlTable(Table tableDefinition) {
     } catch (InvalidConfigurationException | InvalidSchemaException e) {
       throw new InvalidTableException(e.getMessage());
     } catch (JsonProcessingException e) {
-      throw new AssertionError(
-          "Failed to re-parse TBLPROPERTIES JSON " + tableProperties.toString());
+      throw new AssertionError("Failed to re-parse TBLPROPERTIES JSON " + tableProperties);
     }
   }
 

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigquery/BigQueryTable.java

@@ -81,11 +81,11 @@ class BigQueryTable extends SchemaBaseBeamTable implements Serializable {
     this.conversionOptions = options;
     this.bqLocation = table.getLocation();
 
-    if (table.getProperties().containsKey(METHOD_PROPERTY)) {
+    if (table.getProperties().has(METHOD_PROPERTY)) {
       List<String> validMethods =
           Arrays.stream(Method.values()).map(Enum::toString).collect(Collectors.toList());
       // toUpperCase should make it case-insensitive
-      String selectedMethod = table.getProperties().getString(METHOD_PROPERTY).toUpperCase();
+      String selectedMethod = table.getProperties().get(METHOD_PROPERTY).asText().toUpperCase();
 
       if (validMethods.contains(selectedMethod)) {
         method = Method.valueOf(selectedMethod);
@@ -105,12 +105,12 @@ class BigQueryTable extends SchemaBaseBeamTable implements Serializable {
 
     LOG.info("BigQuery method is set to: {}", method);
 
-    if (table.getProperties().containsKey(WRITE_DISPOSITION_PROPERTY)) {
+    if (table.getProperties().has(WRITE_DISPOSITION_PROPERTY)) {
       List<String> validWriteDispositions =
           Arrays.stream(WriteDisposition.values()).map(Enum::toString).collect(Collectors.toList());
       // toUpperCase should make it case-insensitive
       String selectedWriteDisposition =
-          table.getProperties().getString(WRITE_DISPOSITION_PROPERTY).toUpperCase();
+          table.getProperties().get(WRITE_DISPOSITION_PROPERTY).asText().toUpperCase();
 
       if (validWriteDispositions.contains(selectedWriteDisposition)) {
         writeDisposition = WriteDisposition.valueOf(selectedWriteDisposition);

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigquery/BigQueryTableProvider.java

@@ -17,9 +17,7 @@
  */
 package org.apache.beam.sdk.extensions.sql.meta.provider.bigquery;
 
-import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects.firstNonNull;
-
-import com.alibaba.fastjson.JSONObject;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.auto.service.AutoService;
 import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
 import org.apache.beam.sdk.extensions.sql.meta.Table;
@@ -56,10 +54,10 @@ public BeamSqlTable buildBeamSqlTable(Table table) {
     return new BigQueryTable(table, getConversionOptions(table.getProperties()));
   }
 
-  protected static ConversionOptions getConversionOptions(JSONObject properties) {
+  protected static ConversionOptions getConversionOptions(ObjectNode properties) {
     return ConversionOptions.builder()
         .setTruncateTimestamps(
-            firstNonNull(properties.getBoolean("truncateTimestamps"), false)
+            properties.path("truncateTimestamps").asBoolean(false)
                 ? TruncateTimestamps.TRUNCATE
                 : TruncateTimestamps.REJECT)
         .build();

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigtable/BigtableTable.java

@@ -23,7 +23,7 @@
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps.newHashMap;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets.newHashSet;
 
-import com.alibaba.fastjson.JSONObject;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.io.Serializable;
 import java.util.Collection;
 import java.util.HashMap;
@@ -84,9 +84,9 @@ public class BigtableTable extends SchemaBaseBeamTable implements Serializable {
       this.emulatorHost = host;
     }
 
-    JSONObject properties = table.getProperties();
-    if (properties.containsKey(COLUMNS_MAPPING)) {
-      columnsMapping = parseColumnsMapping(properties.getString(COLUMNS_MAPPING));
+    ObjectNode properties = table.getProperties();
+    if (properties.has(COLUMNS_MAPPING)) {
+      columnsMapping = parseColumnsMapping(properties.get(COLUMNS_MAPPING).asText());
       validateColumnsMapping(columnsMapping, schema);
       useFlatSchema = true;
     }