core: add JSON parser for ContentFile and FileScanTask

[stevenzwu]

this closes issue #1698.

There are two motivations as described by issue #1698.

1. provide a more stable serialization (than Java serialization) for Flink checkpoint
2. can be used by REST catalog for scan planning or committing files

[stevenzwu]

this is needed so that FileScanTaskParser (added in this PR) can serialize the schema. Then during the deserialization part, schema can be pass into the constructor of BaseFileScanTask.

Keep it at this level (not base ContentScanTask interface or lower) to limit the scope of change.

[stevenzwu]

exposed as protected so that BaseFileScanTask can use it to implement the FileScanTask#schema() method

[RussellSpitzer]

Little odd that we reverse engineer the schema from the string here, but seems like the most backwards compatible thing we can do here.

[stevenzwu]

agree it is a little odd. On the other hand, partition spec is in the same model in this class. As you said, otherwise we would have to change the constructors of a bunch of classes. The current choice of passing schema and spec as strings is to make those scan tasks serializable.

 @Override
public PartitionSpec spec() {
if (spec == null) {
synchronized (this) {
if (spec == null) {
this.spec = PartitionSpecParser.fromJson(schema(), specString);
}
}
}
return spec;
}

cc @nastra

[stevenzwu]

since DataFile and DeleteFile has the same structure, calling this ContentFileParser without any generic type.

[stevenzwu]

relocated the line here to follow the same order of definition

[stevenzwu]

add a setter for equalityFieldIds so that the parser unit test can cover this field too.

[stevenzwu]

Unlike other JSON parser with a static singleton pattern, ContentFileParser depends on the partition spec. Hence this is a regular class and constructor.

[nastra]

did a high-level pass over the parsers themselves and left a few comments. I haven't had a chance to look closer at the tests yet

[stevenzwu]

@nastra thx a lot for the initial review. I addressed the comments in the latest commit

[nastra]

sorry for the late re-review @stevenzwu, I've left a few more comments.

[nastra]

I've been mainly focusing on the JSON parsers and left a few comments, but overall this looks almost ready. It would be great to get some additional input from another reviewer

[nastra]

I think it would be good to also add a test with a plain JSON string to see how the full JSON looks like. And then maybe also another test with a plain JSON string where all optional fields (metrics, equality field ids, sort order id, split offsets, ...) are missing

[nastra]

nit: maybe we should mention that we're trying to parse this from text to a binary representation

[stevenzwu]

I also fixed a couple other error msgs with the same problem.

[stevenzwu]

Spark CI build failed with some seemingly env problem

 Caused by:
java.lang.RuntimeException: The root scratch dir: /tmp/hive on HDFS should be writable. Current permissions are: rwxr-xr-x
at org.apache.hadoop.hive.ql.session.SessionState.createRootHDFSDir(SessionState.java:724)
at org.apache.hadoop.hive.ql.session.SessionState.createSessionDirs(SessionState.java:654)
at org.apache.hadoop.hive.ql.session.SessionState.start(SessionState.java:586)
at org.apache.hadoop.hive.ql.session.SessionState.start(SessionState.java:548)
at org.apache.spark.sql.hive.client.HiveClientImpl.newState(HiveClientImpl.scala:174)
at org.apache.spark.sql.hive.client.HiveClientImpl.<init>(HiveClientImpl.scala:129)
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:490)
at org.apache.spark.sql.hive.client.IsolatedClientLoader.createClient(IsolatedClientLoader.scala:293)
at org.apache.spark.sql.hive.HiveUtils$.newClientForMetadata(HiveUtils.scala:492)
at org.apache.spark.sql.hive.HiveUtils$.newClientForMetadata(HiveUtils.scala:352)
at org.apache.spark.sql.hive.HiveExternalCatalog.client$lzycompute(HiveExternalCatalog.scala:71)
at org.apache.spark.sql.hive.HiveExternalCatalog.client(HiveExternalCatalog.scala:70)
at org.apache.spark.sql.hive.HiveExternalCatalog.$anonfun$databaseExists$1(HiveExternalCatalog.scala:224)
at scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)
at org.apache.spark.sql.hive.HiveExternalCatalog.withClient(HiveExternalCatalog.scala:102)

[stevenzwu]

merging after rebase

[puchengy]

@stevenzwu we are seeing Trino OOM issue during scan planning and it might be because we introduce table schema to each file scan task in this PR. The issue happens in conjunction with very wide table schema and ParallelIterable usage. I wonder your thought on this? One possible way is to store schema id instead of actual schema to save memory usage.

LMK if this is the right place to discuss or we can move somewhere else.

[stevenzwu]

@puchengy please create a new issue to track and discuss this problem.

Agree with the overhead of serializing the schema for scan task. If we were to just serialize schema id, the serializer would need to get hold of the schemas. It would require major refactoring of the call stack to allow pass-in. At that time, we opted into the simpler approach. But we can discuss the alternative.

Does Trino use the JSON parser for file scan task?