public static HiveWriterFactory getWriterFactory(HiveShim hiveShim, String hiveVersion,
            ObjectIdentifier identifier) {
        if (!CacheHolder.getFactoryMap().containsKey(identifier)) {
            HiveConf hiveConf = HiveTableInlongFactory.getHiveConf();
            try (HiveMetastoreClientWrapper client = HiveMetastoreClientFactory.create(hiveConf, hiveVersion)) {
                List<String> tableNames = client.getAllTables(identifier.getDatabaseName());
                LOG.info("table names: {}", Arrays.deepToString(tableNames.toArray()));
                boolean tableExist = tableNames.stream().anyMatch(identifier.getObjectName()::equalsIgnoreCase);
                if (!tableExist) {
                    return null;
                }

                Table table = client.getTable(identifier.getDatabaseName(), identifier.getObjectName());
                StorageDescriptor sd = table.getSd();

                List<FieldSchema> fieldSchemaList = getTableFields(client, identifier);

                List<FieldSchema> partitionSchemas = table.getPartitionKeys();
                String[] partitions = partitionSchemas.stream().map(FieldSchema::getName).toArray(String[]::new);

                TableSchema.Builder builder = new TableSchema.Builder();
                for (FieldSchema fieldSchema : fieldSchemaList) {
                    LogicalType logicalType = LogicalTypeParser.parse(fieldSchema.getType());
                    if (logicalType instanceof TimestampType) {
                        TimestampType timestampType = (TimestampType) logicalType;
                        logicalType = new TimestampType(timestampType.isNullable(), timestampType.getKind(), 9);
                    } else if (logicalType instanceof ZonedTimestampType) {
                        ZonedTimestampType timestampType = (ZonedTimestampType) logicalType;
                        logicalType = new ZonedTimestampType(timestampType.isNullable(), timestampType.getKind(), 9);
                    }
                    builder.field(fieldSchema.getName(), new AtomicDataType(logicalType));
                }
                TableSchema schema = builder.build();

                Class hiveOutputFormatClz = hiveShim.getHiveOutputFormatClass(Class.forName(sd.getOutputFormat()));
                boolean isCompressed = hiveConf.getBoolean(HiveConf.ConfVars.COMPRESSRESULT.varname, false);
                HiveWriterFactory writerFactory = new HiveWriterFactory(new JobConf(hiveConf), hiveOutputFormatClz, sd,
                        schema, partitions, HiveReflectionUtils.getTableMetadata(hiveShim, table), hiveShim,
                        isCompressed, true);
                CacheHolder.getFactoryMap().put(identifier, writerFactory);
            } catch (TException e) {
                throw new CatalogException("Failed to query Hive metaStore", e);
            } catch (ClassNotFoundException e) {
                throw new FlinkHiveException("Failed to get output format class", e);
            } catch (NoSuchFieldException | IllegalAccessException e) {
                throw new FlinkHiveException("Failed to get hive metastore client", e);
            }
        }
        return CacheHolder.getFactoryMap().get(identifier);
    }
Cache hive wrtier factory

@param hiveShim hiveShim object
@param hiveVersion hive version
@param identifier object identifier
@return hive writer factory
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/util/HiveTableUtil.java