[Bug]: Precision loss for sub fields inside JSON Column in BigQuery

[build_gradle.txt](https://github.com/apache/beam/files/10869011/build_gradle.txt)
### What happened?

Hello,

I have a scenario to build a data pipeline that ingests source data which is in ndjson file (from GCS) to BigQuery.
Here, I want to define a field type as JSON while defining BigQuery Schema.
After successful job run, Subfields inside this particular JSON field are loosing their accuracy. How can I resolve this issue?

As my source data is dynamic in nature, I don't want to explicitly mention subfields schema while defining BigQuery Schema. And I want to write as JSON datatype only(not as String or RECORD etc) without loosing accuracy.
FYI, I have attached build.gradle.kts file

**Source Code:**
```
`package org.apache.beam.examples;
import org.apache.beam.examples.common.ExampleUtils;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Distribution;
import org.apache.beam.sdk.metrics.Metrics;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.transforms.Filter;
import org.apache.beam.sdk.transforms.FlatMapElements;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.JsonSyntaxException;
import java.util.logging.Logger;
import com.google.gson.JsonArray;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.Method;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;

public class json_data {
static final Logger LOG = Logger.getLogger(json_data.class.getName());
static class ExtractData extends DoFn {
static final Logger LOG = Logger.getLogger(ExtractData.class.getName());
@ProcessElement
public void processElement(ProcessContext c) {
String lines = c.element();
String[] lineArray = lines.split("\n");
for (String line : lineArray) {
try {
JsonObject json = new JsonParser().parse(line).getAsJsonObject();
TableRow row = new TableRow()
.set("_index", json.get("_index").getAsString())
.set("_id", json.get("_id").getAsString())
.set("_type", json.get("_type").getAsString())
.set("_score", json.get("_score").getAsString())
.set("_source", json.get("_source").toString());
c.output(row);
} catch (JsonSyntaxException e) {
LOG.severe("Error processing line: " + line + " " + e);
} catch (NullPointerException e) {
LOG.severe("Error processing line: " + line + " " + e);
} catch (JsonProcessingException e) {
LOG.severe("Error processing line: " + line + " " + e);
}

}
}
}
public interface WordCountOptions extends PipelineOptions {
// Set a default input file path
@Description("Path of the file to read from")
@Default.String("inputpathlocation")
String getInputFile();
void setInputFile(String value);

@Description("Path of the file to write to")
@Required
String getOutput();
void setOutput(String value);
}
static void runWordCount(WordCountOptions options) {
Pipeline p = Pipeline.create(options);
PCollection input = p.apply("ReadJSON", TextIO.read().from(options.getInputFile()));
LOG.info("Read JSON from: " + options.getInputFile());
PCollection ParsedJson = input.apply("ParseJson", ParDo.of(new ExtractData()));
LOG.info("Parsed JSON data.");
ParsedJson.apply("Write to BigQuery",
BigQueryIO.writeTableRows().to("BQTableoutptlocation")
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
.withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)
//.withMethod(BigQueryIO.write().withMethod(FILE_LOADS))
//.withMethod(Method.FILE_LOADS)
.withSchema(
new TableSchema().setFields(
Arrays.asList(
new TableFieldSchema().setName("_index").setType("STRING").setMode("NULLABLE"),
new TableFieldSchema().setName("_id").setType("STRING").setMode("NULLABLE"),
new TableFieldSchema().setName("_type").setType("STRING").setMode("NULLABLE"),
new TableFieldSchema().setName("_score").setType("STRING").setMode("NULLABLE"),
new TableFieldSchema().setName("_source").setType("JSON").setMode("NULLABLE")
)
)
));
p.run().waitUntilFinish();
}

public static void main(String[] args) {
WordCountOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(WordCountOptions.class);
runWordCount(options);
}
}`

```
**Gradle command for execution(Using Direct runner):**

gradle clean execute -DmainClass=org.apache.beam.examples.json_data --args="--project=name_of_the_gcpproject --output=project_name:datasetname.tablename --stagingLocation=gs:// --tempLocation=gs:// --region=us-central1" -Pdataflow-runner

**Approach 2:**
We have tried ingesting JSON messages from pub/sub topic to BigQuery table (as JSON datatype only). We used Apache Beam Python SDK for dataflow job in this approach and still encountered this precision loss in BigQuery JSON field.

**Source Code:**
```
import apache_beam as beam
from apache_beam.options.pipeline_options import GoogleCloudOptions
import google.auth
import json
from apache_beam.options import pipeline_options
options = pipeline_options.PipelineOptions()
options.view_as(pipeline_options.StandardOptions).streaming = True

schema = 'event:JSON'
p = beam.Pipeline( options=options)
events = p | "read" >> beam.io.ReadFromPubSub(topic="projects/project_name/topics/tipic_name")
windowed_events = (events
| 'to json' >> beam.Map(lambda x: json.loads(x.decode('utf-8')))
| 'represent as TableRow json (column_name: value)' >> beam.Map(lambda x: "{\"event\": %s}" %json.dumps(json.dumps(x)))
| 'back to json' >> beam.Map(lambda x: json.loads(x))
| 'Write to BQ' >> beam.io.WriteToBigQuery(
'projectname.dataset_name.tablename',
schema=schema,
ignore_unknown_columns=True,
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
)
p.run()
```

### Issue Priority

Priority: 3 (minor)

### Issue Components

- [x] Component: Python SDK
- [X] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [X] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner