@Override
    public LinkedHashMap<String, String> generatePartValues(RowData in) {
        LinkedHashMap<String, String> partSpec = new LinkedHashMap<>();

        if (sinkMultipleEnable) {
            GenericRowData rowData = (GenericRowData) in;
            JsonNode rootNode;
            try {
                if (jsonFormat == null) {
                    jsonFormat = (JsonDynamicSchemaFormat) DynamicSchemaFormatFactory.getFormat(sinkMultipleFormat);
                }
                rootNode = jsonFormat.deserialize((byte[]) rowData.getField(0));

                String databaseName = jsonFormat.parse(rootNode, databasePattern);
                String tableName = jsonFormat.parse(rootNode, tablePattern);
                List<Map<String, Object>> physicalDataList = HiveTableUtil.jsonNode2Map(
                        jsonFormat.getPhysicalData(rootNode));

                List<String> pkListStr = jsonFormat.extractPrimaryKeyNames(rootNode);
                RowType schema = jsonFormat.extractSchema(rootNode, pkListStr);

                Map<String, Object> rawData = physicalDataList.get(0);
                ObjectIdentifier identifier = HiveTableUtil.createObjectIdentifier(databaseName, tableName);

                HashMap<ObjectIdentifier, Long> ignoreWritingTableMap = CacheHolder.getIgnoreWritingTableMap();
                // ignore writing data into this table
                if (ignoreWritingTableMap.containsKey(identifier)) {
                    return partSpec;
                }

                HiveWriterFactory hiveWriterFactory = HiveTableUtil.getWriterFactory(hiveShim, hiveVersion, identifier);
                if (hiveWriterFactory == null) {
                    HiveTableUtil.createTable(databaseName, tableName, schema, partitionPolicy, hiveVersion,
                            inputFormat, outputFormat, serializationLib);
                    hiveWriterFactory = HiveTableUtil.getWriterFactory(hiveShim, hiveVersion, identifier);
                }

                String[] partitionColumns = hiveWriterFactory.getPartitionColumns();
                if (partitionColumns.length == 0) {
                    // non partition table
                    return partSpec;
                }

                String[] columnNames = hiveWriterFactory.getAllColumns();
                DataType[] allTypes = hiveWriterFactory.getAllTypes();
                List<String> columnList = Arrays.asList(columnNames);
                int[] partitionIndexes = Arrays.stream(partitionColumns).mapToInt(columnList::indexOf).toArray();

                boolean replaceLineBreak = hiveWriterFactory.getStorageDescriptor().getInputFormat()
                        .contains("TextInputFormat");
                Pair<GenericRowData, Integer> pair = HiveTableUtil.getRowData(rawData, columnNames, allTypes,
                        replaceLineBreak);
                GenericRowData genericRowData = pair.getLeft();

                Object field;
                for (int i = 0; i < partitionIndexes.length; i++) {
                    int fieldIndex = partitionIndexes[i];
                    field = genericRowData.getField(fieldIndex);
                    DataType type = allTypes[fieldIndex];
                    ObjectInspector objectInspector = HiveInspectors.getObjectInspector(type);
                    HiveObjectConversion hiveConversion = HiveInspectors.getConversion(objectInspector,
                            type.getLogicalType(), hiveShim);
                    String partitionValue = field != null ? String.valueOf(hiveConversion.toHiveObject(field)) : null;
                    if (partitionValue == null) {
                        field = HiveTableUtil.getDefaultPartitionValue(rawData, schema, partitionPolicy, partitionField,
                                timePattern);
                        partitionValue = field != null ? String.valueOf(hiveConversion.toHiveObject(field)) : null;
                    }
                    if (StringUtils.isEmpty(partitionValue)) {
                        partitionValue = defaultPartValue;
                    }
                    partSpec.put(partitionColumns[i], partitionValue);
                }
            } catch (Exception e) {
                // do not throw exception, just log it. so HadoopPathBasedPartFileWriter will archive dirty data or not
                LOG.error("Generate partition values error", e);
            }
        } else {
            for (int i = 0; i < partitionIndexes.length; i++) {
                Object field = partitionConverters[i].toExternal(in, partitionIndexes[i]);
                String partitionValue = field != null ? hiveObjectConversions[i].toHiveObject(field).toString() : null;
                if (StringUtils.isEmpty(partitionValue)) {
                    partitionValue = defaultPartValue;
                }
                partSpec.put(partitionColumns[i], partitionValue);
            }
        }
        return partSpec;
    }
A {@link RowDataPartitionComputer} that converts Flink objects to Hive objects before computing
the partition value strings.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/HiveRowDataPartitionComputer.java