private static void setupHoodieKeyOptions(Configuration conf, CatalogTable table) {
        List<String> pkColumns = table.getSchema().getPrimaryKey()
                .map(UniqueConstraint::getColumns).orElse(Collections.emptyList());
        if (pkColumns.size() > 0) {
            // the PRIMARY KEY syntax always has higher priority than option FlinkOptions#RECORD_KEY_FIELD
            String recordKey = String.join(",", pkColumns);
            conf.setString(FlinkOptions.RECORD_KEY_FIELD, recordKey);
        }
        List<String> partitionKeys = table.getPartitionKeys();
        if (partitionKeys.size() > 0) {
            // the PARTITIONED BY syntax always has higher priority than option FlinkOptions#PARTITION_PATH_FIELD
            conf.setString(FlinkOptions.PARTITION_PATH_FIELD, String.join(",", partitionKeys));
        }
        // set index key for bucket index if not defined
        if (conf.getString(FlinkOptions.INDEX_TYPE).equals(HoodieIndex.IndexType.BUCKET.name())) {
            if (conf.getString(FlinkOptions.INDEX_KEY_FIELD).isEmpty()) {
                conf.setString(FlinkOptions.INDEX_KEY_FIELD, conf.getString(FlinkOptions.RECORD_KEY_FIELD));
            } else {
                Set<String> recordKeySet =
                        Arrays.stream(conf.getString(FlinkOptions.RECORD_KEY_FIELD).split(","))
                                .collect(Collectors.toSet());
                Set<String> indexKeySet =
                        Arrays.stream(conf.getString(FlinkOptions.INDEX_KEY_FIELD).split(","))
                                .collect(Collectors.toSet());
                if (!recordKeySet.containsAll(indexKeySet)) {
                    throw new HoodieValidationException(
                            FlinkOptions.INDEX_KEY_FIELD + " should be a subset of or equal to the recordKey fields");
                }
            }
        }

        // tweak the key gen class if possible
        final String[] partitions = conf.getString(FlinkOptions.PARTITION_PATH_FIELD).split(",");
        final String[] pks = conf.getString(FlinkOptions.RECORD_KEY_FIELD).split(",");
        if (partitions.length == 1) {
            final String partitionField = partitions[0];
            if (partitionField.isEmpty()) {
                conf.setString(FlinkOptions.KEYGEN_CLASS_NAME, NonpartitionedAvroKeyGenerator.class.getName());
                LOG.info("Table option [{}] is reset to {} because this is a non-partitioned table",
                        FlinkOptions.KEYGEN_CLASS_NAME.key(), NonpartitionedAvroKeyGenerator.class.getName());
                return;
            }
            DataType partitionFieldType = table.getSchema().getFieldDataType(partitionField)
                    .orElseThrow(() -> new HoodieValidationException("Field " + partitionField + " does not exist"));
            if (pks.length <= 1 && DataTypeUtils.isDatetimeType(partitionFieldType)) {
                // timestamp based key gen only supports simple primary key
                setupTimestampKeygenOptions(conf, partitionFieldType);
                return;
            }
        }
        boolean complexHoodieKey = pks.length > 1 || partitions.length > 1;
        if (complexHoodieKey && FlinkOptions.isDefaultValueDefined(conf, FlinkOptions.KEYGEN_CLASS_NAME)) {
            conf.setString(FlinkOptions.KEYGEN_CLASS_NAME, ComplexAvroKeyGenerator.class.getName());
            LOG.info("Table option [{}] is reset to {} because record key or partition path has two or more fields",
                    FlinkOptions.KEYGEN_CLASS_NAME.key(), ComplexAvroKeyGenerator.class.getName());
        }
    }
Sets up the hoodie key options (e.g. record key and partition key) from the table definition.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/table/HoodieTableFactory.java