private void sanityCheck(Configuration conf, ResolvedSchema schema) {
        List<String> fields = schema.getColumnNames();

        // validate record key in pk absence.
        if (!schema.getPrimaryKey().isPresent()) {
            String[] recordKeys = conf.get(FlinkOptions.RECORD_KEY_FIELD).split(",");
            if (recordKeys.length == 1
                    && FlinkOptions.RECORD_KEY_FIELD.defaultValue().equals(recordKeys[0])
                    && !fields.contains(recordKeys[0])) {
                throw new HoodieValidationException("Primary key definition is required, use either PRIMARY KEY syntax "
                        + "or option '" + FlinkOptions.RECORD_KEY_FIELD.key() + "' to specify.");
            }

            Arrays.stream(recordKeys)
                    .filter(field -> !fields.contains(field))
                    .findAny()
                    .ifPresent(f -> {
                        throw new HoodieValidationException("Field '" + f + "' specified in option "
                                + "'" + FlinkOptions.RECORD_KEY_FIELD.key() + "' does not exist in the table schema.");
                    });
        }

        // validate pre_combine key
        String preCombineField = conf.get(FlinkOptions.PRECOMBINE_FIELD);
        if (!fields.contains(preCombineField)) {
            if (OptionsResolver.isDefaultHoodieRecordPayloadClazz(conf)) {
                throw new HoodieValidationException("Option '" + FlinkOptions.PRECOMBINE_FIELD.key()
                        + "' is required for payload class: " + DefaultHoodieRecordPayload.class.getName());
            }
            if (preCombineField.equals(FlinkOptions.PRECOMBINE_FIELD.defaultValue())) {
                conf.setString(FlinkOptions.PRECOMBINE_FIELD, FlinkOptions.NO_PRE_COMBINE);
            } else if (!preCombineField.equals(FlinkOptions.NO_PRE_COMBINE)) {
                throw new HoodieValidationException("Field " + preCombineField + " does not exist in the table schema."
                        + "Please check '" + FlinkOptions.PRECOMBINE_FIELD.key() + "' option.");
            }
        }
    }
The sanity check.

@param conf   The table options
@param schema The table schema
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/table/HoodieTableFactory.java