[HUDI-8841] Fix schema validating exception during flink async cluste…

[cshuo]

…ring

Change Logs

In Flink SQL, primary key constraint can be defined, and the type of pk field become non-null，while spark has no primary key constraint, so they have discrepancies in the schema of the underlying files, e.g., parquet.

we can make a schema reconciliation during the async clustering reading.

Impact

Consolidating flink clustering by reconcile schemas to tolerate different nullabilities of record key.

Risk level (write none, low medium or high below)

low

Documentation Update

none

Contributor's checklist

• Read through contributor's guide
• Change Logs and Impact were stated clearly
• Adequate tests were added if applicable
• CI passed

[danny0405]

Currently, the Flink clustering only work on append only table, all the data files should be in parquet format, we can fetch the record key fields, then the file schema from the parquet footer, and reconcile the record key fields if the nullability discrepency exists.

[cshuo]

As for where to get the file schema, actually, files with same commit time share the same write schema, and getTableAvroSchema in TableSchemaResolver is more efficient since there is a cache, what do you think

[danny0405]

No, the TableSchemaResolver would trigger file lising of the commit metadata which is a pressure to filesystem. Fetch the schema from file is straight-forward:

private Schema fetchSchemaFromFiles(Iterator<String> filePaths) throws IOException {
    Schema schema = null;
    while (filePaths.hasNext() && schema == null) {
      StoragePath filePath = new StoragePath(filePaths.next());
      if (FSUtils.isLogFile(filePath)) {
        // this is a log file
        schema = readSchemaFromLogFile(filePath);
      } else {
        schema = HoodieIOFactory.getIOFactory(metaClient.getStorage())
            .getFileFormatUtils(filePath).readAvroSchema(metaClient.getStorage(), filePath);
      }
    }
    return schema;
  }

Another way is to force make the record key fields nullable if they are not if a nullable schema can be used to read non-nullable schema data file.

[cshuo]

The second way sounds more efficient, I'll verify it.

[cshuo]

cc @danny0405 updated using the second way, and added a test to verify nullable schema can be used to read non-nullable schema data file

[danny0405]

Rename to forceNullableColumns, we should also avoid to recreate the schema if the field is already nullable.

[danny0405]

The default record key field is uuid, we should check the validity first.

[danny0405]

nullableUpdateCols -> columns. We can eliminate the flag nullalbe because it is always true.

[hudi-bot]

CI report:

• a77e69e Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build