record = generateRecord();
+ TapInsertRecordEvent event = new TapInsertRecordEvent();
+ event.setAfter(record);
+ event.setTableId(tableName);
+ event.setReferenceTime(System.currentTimeMillis());
+ events.add(event);
+ }
+
+ // 调用消费者处理数据
+ consumer.accept(events);
+ totalWritten += events.size();
+
+ // 控制QPS
+ if (qps > 0) {
+ long expectedTimePerBatch = (batchSize * 1000) / qps;
+ long actualTime = System.currentTimeMillis() - startTime;
+ long sleepTime = expectedTimePerBatch - actualTime;
+
+ if (sleepTime > 0) {
+ try {
+ Thread.sleep(sleepTime);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break;
+ }
+ }
+ startTime = System.currentTimeMillis();
+ }
+
+ // 每10批次打印进度
+ if ((i + 1) % 10 == 0 || i == totalBatches - 1) {
+ double progress = (double) (i + 1) / totalBatches * 100;
+ System.out.printf(" >> 进度: %d/%d 批次 (%.1f%%), 已写入: %d 条%n",
+ i + 1, totalBatches, progress, totalWritten);
+ }
+ }
+
+ System.out.println(" >> 数据生成完成: 总计 " + totalWritten + " 条");
+ }
+
+ /**
+ * 获取已生成的唯一ID数量
+ *
+ * 返回实际生成的唯一 ID 总数（非窗口限制值）
+ */
+ public long getUniqueIdsCount() {
+ return idGenerator.get() - 1;
+ }
+
+ /**
+ * 获取总生成记录数(包括重复)
+ */
+ public long getTotalGenerated() {
+ return idGenerator.get() - 1;
+ }
+}
diff --git a/connectors/connector-perf-test/src/main/java/io/tapdata/connector/paimon/perf/PaimonDataGenerator.java b/connectors/connector-perf-test/src/main/java/io/tapdata/connector/paimon/perf/PaimonDataGenerator.java
new file mode 100644
index 000000000..2bf159bbd
--- /dev/null
+++ b/connectors/connector-perf-test/src/main/java/io/tapdata/connector/paimon/perf/PaimonDataGenerator.java
@@ -0,0 +1,130 @@
+package io.tapdata.connector.paimon.perf;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.DataTypes;
+
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * 数据生成器 - 为 AutoTestRunner 专用
+ * 生成交易明细表测试数据（29个字段）
+ */
+public class PaimonDataGenerator {
+ private final long totalRecords;
+ private final int warmupRecords;
+ private final double duplicateRate;
+ private final int batchSize;
+
+ private final Random random;
+ private final AtomicLong generatedCount;
+ private final AtomicLong idCounter;
+
+ // 交易明细表字段数据
+ private static final String[] COMPOR_IDS = {"C001", "C002", "C003", "C004", "C005"};
+ private static final String[] TXN_TYPES = {"SALE", "REFUND", "VOID", "EXCHANGE"};
+ private static final String[] CHANNELS = {"CASH", "CARD", "MOBILE", "ONLINE"};
+ private static final String[] OUTLETS = {"OUTLET_A", "OUTLET_B", "OUTLET_C", "OUTLET_D"};
+ private static final String[] SOURCES = {"POS", "ECOM", "MOBILE_APP"};
+ private static final String[] DOLLAR_TYPES = {"USD", "EUR", "CNY", "JPY"};
+ private static final String[] PROPERTIES = {"NORMAL", "VIP", "PROMO"};
+
+ public PaimonDataGenerator(long totalRecords, int warmupRecords, double duplicateRate, int batchSize) {
+ this.totalRecords = totalRecords;
+ this.warmupRecords = warmupRecords;
+ this.duplicateRate = duplicateRate;
+ this.batchSize = batchSize;
+ this.random = new Random(42); // 固定种子以保证可重复性
+ this.generatedCount = new AtomicLong(0);
+ this.idCounter = new AtomicLong(1);
+ }
+
+ /**
+ * 是否还有更多记录
+ */
+ public boolean hasMore() {
+ return generatedCount.get() < totalRecords;
+ }
+
+ /**
+ * 生成下一条记录
+ */
+ public InternalRow nextRecord() {
+ if (!hasMore()) {
+ return null;
+ }
+
+ long currentId = generatedCount.incrementAndGet();
+
+ // 根据重复率决定是否生成重复ID
+ long id;
+ if (duplicateRate > 0 && generatedCount.get() > warmupRecords && random.nextDouble() < duplicateRate) {
+ id = Math.max(1, idCounter.get() - random.nextInt(1000) - 1);
+ } else {
+ id = idCounter.getAndIncrement();
+ }
+
+ return createRow(id);
+ }
+
+ /**
+ * 创建一行数据
+ */
+ private GenericRow createRow(long id) {
+ GenericRow row = GenericRow.of(
+ id, // id (BIGINT)
+ BinaryString.fromString("BD_" + id), // balance_detail_id (STRING)
+ Decimal.fromBigDecimal(randomBigDecimal(18, 0), 18, 0), // before_detail_balance (DECIMAL)
+ Decimal.fromBigDecimal(randomBigDecimal(18, 0), 18, 0), // amount (DECIMAL)
+ BinaryString.fromString(randomDateTime()), // expiry_date (TIMESTAMP)
+ BinaryString.fromString(randomArrayElement(COMPOR_IDS)), // compor_id (STRING)
+ BinaryString.fromString(randomArrayElement(TXN_TYPES)), // transaction_type (STRING)
+ BinaryString.fromString(randomArrayElement(CHANNELS)), // channel (STRING)
+ BinaryString.fromString("POS_REF_" + id), // pos_reference (STRING)
+ BinaryString.fromString(randomArrayElement(OUTLETS)), // outlet (STRING)
+ BinaryString.fromString("Remark for " + id), // remark (STRING)
+ BinaryString.fromString(randomDateTime()), // created_time (TIMESTAMP)
+ BinaryString.fromString("PD_" + id), // payment_detail_id (STRING)
+ id + 1000000L, // payment_id (BIGINT)
+ BinaryString.fromString("admin"), // created_by (STRING)
+ BinaryString.fromString("I"), // op (STRING)
+ Decimal.fromBigDecimal(randomBigDecimal(18, 0), 18, 0), // after_detail_balance (DECIMAL)
+ BinaryString.fromString(randomArrayElement(SOURCES)), // source_system (STRING)
+ BinaryString.fromString(randomArrayElement(DOLLAR_TYPES)), // dollar_type_id (STRING)
+ Decimal.fromBigDecimal(randomBigDecimal(18, 0), 18, 0), // exception_balance (DECIMAL)
+ BinaryString.fromString("PATRON_" + (id % 1000)), // patron_id (STRING)
+ BinaryString.fromString("KEY_" + id), // source_key (STRING)
+ BinaryString.fromString("DEV_" + (id % 100)), // device_id (STRING)
+ Decimal.fromBigDecimal(randomBigDecimal(18, 0), 18, 0), // after_balance (DECIMAL)
+ Decimal.fromBigDecimal(randomBigDecimal(18, 0), 18, 0), // before_balance (DECIMAL)
+ BinaryString.fromString(randomArrayElement(OUTLETS)), // outlet_code (STRING)
+ BinaryString.fromString(randomDateTime()), // ods_updated_at (TIMESTAMP)
+ BinaryString.fromString(randomArrayElement(PROPERTIES)), // property (STRING)
+ 20240101 + (int)(id % 10000) // pt_created_date (INT)
+ );
+
+ return row;
+ }
+
+ private BigDecimal randomBigDecimal(int precision, int scale) {
+ long maxValue = (long) Math.pow(10, precision - scale) - 1;
+ long randomValue = (long) (random.nextDouble() * maxValue);
+ return BigDecimal.valueOf(randomValue, scale);
+ }
+
+ private String randomDateTime() {
+ long epochDay = 19000 + random.nextInt(1000); // 2022-2025
+ int secondOfDay = random.nextInt(86400);
+ LocalDateTime ldt = LocalDateTime.ofEpochSecond(epochDay * 86400L + secondOfDay, 0, java.time.ZoneOffset.UTC);
+ return ldt.toString();
+ }
+
+ private T randomArrayElement(T[] array) {
+ return array[random.nextInt(array.length)];
+ }
+}
diff --git a/connectors/connector-perf-test/src/main/java/io/tapdata/connector/paimon/perf/PaimonFileObserver.java b/connectors/connector-perf-test/src/main/java/io/tapdata/connector/paimon/perf/PaimonFileObserver.java
new file mode 100644
index 000000000..e94add508
--- /dev/null
+++ b/connectors/connector-perf-test/src/main/java/io/tapdata/connector/paimon/perf/PaimonFileObserver.java
@@ -0,0 +1,380 @@
+package io.tapdata.connector.paimon.perf;
+
+import org.apache.hadoop.fs.FileStatus;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Paimon 文件观测器 - 增强版（支持本地和 S3）
+ * 自动扫描并输出Paimon数据目录的文件列表、文件大小、文件数量
+ */
+public class PaimonFileObserver {
+ private final String warehousePath;
+ private final String database;
+ private final String tableName;
+ private final boolean isS3;
+ private final UnifiedFileSystem fileSystem;
+ private List lastScanFiles = new ArrayList<>();
+
+ /**
+ * 创建本地文件观测器
+ */
+ public PaimonFileObserver(String warehousePath, String database, String tableName) {
+ this.warehousePath = warehousePath;
+ this.database = database;
+ this.tableName = tableName;
+ this.isS3 = false;
+ this.fileSystem = UnifiedFileSystem.createLocal();
+ }
+
+ /**
+ * 创建 S3 文件观测器
+ */
+ public PaimonFileObserver(String warehousePath, String database, String tableName,
+ String s3Endpoint, String s3AccessKey, String s3SecretKey, String s3Region) {
+ this.warehousePath = warehousePath;
+ this.database = database;
+ this.tableName = tableName;
+ this.isS3 = true;
+ // 从仓库路径中提取 bucket 名称 (格式: s3://bucket/key)
+ String bucket = "default-bucket";
+ if (warehousePath.startsWith("s3://")) {
+ String pathWithoutScheme = warehousePath.substring(5);
+ int slashIdx = pathWithoutScheme.indexOf('/');
+ if (slashIdx > 0) {
+ bucket = pathWithoutScheme.substring(0, slashIdx);
+ } else {
+ bucket = pathWithoutScheme;
+ }
+ }
+ this.fileSystem = UnifiedFileSystem.createS3(s3Endpoint, s3AccessKey, s3SecretKey, s3Region, bucket);
+ }
+
+ /**
+ * 获取表的数据目录路径（兼容 bucket-N 和 data/ 两种布局）
+ */
+ public String getTableDataPath() {
+ // Paimon 1.x 动态分桶下数据直接在表根目录下的 bucket-N/ 中
+ return getTablePath();
+ }
+
+ /**
+ * 获取表的完整目录路径
+ */
+ public String getTablePath() {
+ if (isS3) {
+ // S3 路径格式：s3a://bucket/prefix/TC-01/default.db/test_table
+ return warehousePath + "/" + database + ".db" + "/" + tableName;
+ } else {
+ return warehousePath + java.io.File.separator + database + ".db" + java.io.File.separator + tableName;
+ }
+ }
+
+ /**
+ * 扫描表的数据文件（支持 bucket-N/ 和 data/ 两种目录结构）
+ */
+ public List scanDataFiles() throws IOException {
+ List fileInfos = new ArrayList<>();
+ String tablePath = getTablePath();
+
+ try {
+ if (!fileSystem.exists(tablePath) || !fileSystem.isDirectory(tablePath)) {
+ lastScanFiles = fileInfos;
+ return fileInfos;
+ }
+
+ // 递归列出所有文件
+ List fileStatuses = fileSystem.listFilesRecursive(tablePath);
+
+ for (FileStatus status : fileStatuses) {
+ String fileName = status.getPath().getName();
+ String pathStr = status.getPath().toString();
+
+ // 只统计数据文件（parquet/orc/avro），排除元数据
+ if ((fileName.endsWith(".parquet") ||
+ fileName.endsWith(".orc") ||
+ fileName.endsWith(".avro"))
+ && !pathStr.contains("/snapshot/")
+ && !pathStr.contains("/schema/")
+ && !pathStr.contains("/index/")
+ && !pathStr.contains("/manifest/")) {
+ fileInfos.add(new FileInfo(
+ status.getPath().toString(),
+ status.getLen(),
+ status.getModificationTime()
+ ));
+ }
+ }
+ } catch (Exception e) {
+ System.err.println(" [WARN] 扫描文件失败: " + e.getMessage());
+ PerformanceTestRunner.printStackTrace(e);
+ }
+
+ lastScanFiles = fileInfos;
+ return fileInfos;
+ }
+
+ /**
+ * 打印详细文件信息
+ */
+ public void printFileInfo() throws IOException {
+ List fileInfos = scanDataFiles();
+ printFileInfo(fileInfos);
+ }
+
+ /**
+ * 打印文件信息(使用已扫描的结果)
+ */
+ public void printFileInfo(List fileInfos) throws IOException {
+ System.out.println("\n" + "=".repeat(70));
+ System.out.println(" Paimon 数据文件观测报告");
+ if (isS3) {
+ System.out.println(" 存储类型: S3 对象存储");
+ } else {
+ System.out.println(" 存储类型: 本地文件系统");
+ }
+ System.out.println("=".repeat(70));
+ System.out.println("表路径: " + getTableDataPath());
+ System.out.println("扫描时间: " + new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
+ System.out.println("-".repeat(70));
+
+ if (fileInfos.isEmpty()) {
+ System.out.println("⚠ 未找到数据文件!");
+ System.out.println("=".repeat(70));
+ return;
+ }
+
+ // 基础统计
+ long totalSize = fileInfos.stream().mapToLong(FileInfo::getSize).sum();
+ long minSize = fileInfos.stream().mapToLong(FileInfo::getSize).min().orElse(0);
+ long maxSize = fileInfos.stream().mapToLong(FileInfo::getSize).max().orElse(0);
+ double avgSize = (double) totalSize / fileInfos.size();
+
+ System.out.println("📊 文件统计:");
+ System.out.println(" 文件总数: " + fileInfos.size() + " 个");
+ System.out.println(" 总大小: " + formatSize(totalSize));
+ System.out.println(" 平均大小: " + formatSize((long) avgSize));
+ System.out.println(" 最小文件: " + formatSize(minSize));
+ System.out.println(" 最大文件: " + formatSize(maxSize));
+
+ // 文件大小分布
+ System.out.println("\n📏 文件大小分布:");
+ Map sizeDistribution = calculateSizeDistribution(fileInfos);
+ for (Map.Entry entry : sizeDistribution.entrySet()) {
+ String bar = "█".repeat(Math.max(1, entry.getValue().intValue()));
+ System.out.printf(" %-20s: %3d 个 %s%n", entry.getKey(), entry.getValue(), bar);
+ }
+
+ // 文件列表(前20个)
+ System.out.println("\n📄 文件列表(前20个):");
+ System.out.printf(" %-70s %12s%n", "文件路径", "大小");
+ System.out.println(" " + "-".repeat(70) + " " + "-".repeat(12));
+
+ int count = 0;
+ String tablePath = getTableDataPath();
+ for (FileInfo fileInfo : fileInfos) {
+ if (count >= 20) break;
+ // 显示相对路径
+ String relativePath = fileInfo.getPath();
+ if (relativePath.startsWith(tablePath)) {
+ relativePath = relativePath.substring(tablePath.length() + 1);
+ }
+ System.out.printf(" %-70s %12s%n", relativePath, formatSize(fileInfo.getSize()));
+ count++;
+ }
+
+ if (fileInfos.size() > 20) {
+ System.out.println(" ... 还有 " + (fileInfos.size() - 20) + " 个文件");
+ }
+
+ System.out.println("=".repeat(70));
+ }
+
+ /**
+ * 计算文件大小分布
+ */
+ private Map calculateSizeDistribution(List fileInfos) {
+ Map distribution = new LinkedHashMap<>();
+ distribution.put("< 1KB", fileInfos.stream().filter(f -> f.getSize() < 1024).count());
+ distribution.put("1KB - 1MB", fileInfos.stream().filter(f ->
+ f.getSize() >= 1024 && f.getSize() < 1024 * 1024).count());
+ distribution.put("1MB - 10MB", fileInfos.stream().filter(f ->
+ f.getSize() >= 1024 * 1024 && f.getSize() < 10 * 1024 * 1024).count());
+ distribution.put("10MB - 100MB", fileInfos.stream().filter(f ->
+ f.getSize() >= 10 * 1024 * 1024 && f.getSize() < 100 * 1024 * 1024).count());
+ distribution.put("100MB - 500MB", fileInfos.stream().filter(f ->
+ f.getSize() >= 100 * 1024 * 1024 && f.getSize() < 500 * 1024 * 1024).count());
+ distribution.put("> 500MB", fileInfos.stream().filter(f ->
+ f.getSize() >= 500 * 1024 * 1024).count());
+ return distribution;
+ }
+
+ /**
+ * 监控文件变化
+ */
+ public void monitorFileChanges(long durationMs) throws IOException, InterruptedException {
+ System.out.println("\n" + "=".repeat(70));
+ System.out.println(" 开始监控文件变化 (持续时间: " + (durationMs / 1000) + "秒)");
+ System.out.println("=".repeat(70));
+
+ List initialFiles = scanDataFiles();
+ System.out.println("初始文件数量: " + initialFiles.size());
+ System.out.println("初始总大小: " + formatSize(initialFiles.stream().mapToLong(FileInfo::getSize).sum()));
+
+ long startTime = System.currentTimeMillis();
+ int checkCount = 0;
+
+ while (System.currentTimeMillis() - startTime < durationMs) {
+ TimeUnit.SECONDS.sleep(2);
+ List currentFiles = scanDataFiles();
+ long currentSize = currentFiles.stream().mapToLong(FileInfo::getSize).sum();
+ checkCount++;
+
+ System.out.printf(" [%ds] 检查#%d: 文件数=%d, 总大小=%s, 新增文件=%d%n",
+ (System.currentTimeMillis() - startTime) / 1000,
+ checkCount,
+ currentFiles.size(),
+ formatSize(currentSize),
+ currentFiles.size() - initialFiles.size());
+
+ initialFiles = currentFiles;
+ }
+
+ System.out.println("监控结束");
+ }
+
+ /**
+ * 对比两次扫描的差异
+ */
+ public FileChangeReport compareWithLastScan() throws IOException {
+ List currentFiles = scanDataFiles();
+ FileChangeReport report = new FileChangeReport();
+
+ report.setPreviousFileCount(lastScanFiles.size());
+ report.setCurrentFileCount(currentFiles.size());
+ report.setNewFiles(currentFiles.size() - lastScanFiles.size());
+
+ long previousSize = lastScanFiles.stream().mapToLong(FileInfo::getSize).sum();
+ long currentSize = currentFiles.stream().mapToLong(FileInfo::getSize).sum();
+ report.setPreviousSize(previousSize);
+ report.setCurrentSize(currentSize);
+ report.setSizeChange(currentSize - previousSize);
+
+ return report;
+ }
+
+ /**
+ * 格式化文件大小
+ */
+ public static String formatSize(long size) {
+ if (size < 1024) {
+ return size + " B";
+ } else if (size < 1024 * 1024) {
+ return String.format("%.2f KB", (double) size / 1024);
+ } else if (size < 1024 * 1024 * 1024) {
+ return String.format("%.2f MB", (double) size / (1024 * 1024));
+ } else {
+ return String.format("%.2f GB", (double) size / (1024 * 1024 * 1024));
+ }
+ }
+
+ /**
+ * 文件信息类
+ */
+ public static class FileInfo {
+ private final String path;
+ private final long size;
+ private final long lastModified;
+
+ public FileInfo(String path, long size, long lastModified) {
+ this.path = path;
+ this.size = size;
+ this.lastModified = lastModified;
+ }
+
+ public String getPath() { return path; }
+ public long getSize() { return size; }
+ public long getLastModified() { return lastModified; }
+
+ @Override
+ public String toString() {
+ return String.format("FileInfo{path='%s', size=%s, lastModified=%d}",
+ path, formatSize(size), lastModified);
+ }
+ }
+
+ /**
+ * 文件变化报告
+ */
+ public static class FileChangeReport {
+ private int previousFileCount;
+ private int currentFileCount;
+ private int newFiles;
+ private long previousSize;
+ private long currentSize;
+ private long sizeChange;
+
+ public int getPreviousFileCount() { return previousFileCount; }
+ public void setPreviousFileCount(int previousFileCount) { this.previousFileCount = previousFileCount; }
+ public int getCurrentFileCount() { return currentFileCount; }
+ public void setCurrentFileCount(int currentFileCount) { this.currentFileCount = currentFileCount; }
+ public int getNewFiles() { return newFiles; }
+ public void setNewFiles(int newFiles) { this.newFiles = newFiles; }
+ public long getPreviousSize() { return previousSize; }
+ public void setPreviousSize(long previousSize) { this.previousSize = previousSize; }
+ public long getCurrentSize() { return currentSize; }
+ public void setCurrentSize(long currentSize) { this.currentSize = currentSize; }
+ public long getSizeChange() { return sizeChange; }
+ public void setSizeChange(long sizeChange) { this.sizeChange = sizeChange; }
+
+ @Override
+ public String toString() {
+ return String.format("文件变化: 数量 %d -> %d (新增%d), 大小 %s -> %s (变化%s)",
+ previousFileCount, currentFileCount, newFiles,
+ formatSize(previousSize), formatSize(currentSize),
+ (sizeChange >= 0 ? "+" : "") + formatSize(sizeChange));
+ }
+ }
+
+ /**
+ * 扫描所有文件（包含 data 目录和其他元数据目录下的数据文件）
+ */
+ public List scanAllFiles() throws IOException {
+ return scanDataFiles();
+ }
+
+ /**
+ * 紧凑格式打印（至15个文件）
+ */
+ public void printCompact() throws IOException {
+ List files = scanDataFiles();
+ if (files.isEmpty()) {
+ System.out.println(" [文件] 暂无数据文件（可能尚未 flush）");
+ return;
+ }
+ long total = files.stream().mapToLong(FileInfo::getSize).sum();
+ System.out.printf(" [文件] 数量: %d 总大小: %s%n", files.size(), formatSize(total));
+ int shown = Math.min(files.size(), 15);
+ String tablePath = getTablePath();
+ for (int i = 0; i < shown; i++) {
+ FileInfo fi = files.get(i);
+ String rel = fi.getPath();
+ if (rel.startsWith(tablePath)) {
+ rel = rel.substring(tablePath.length() + 1);
+ }
+ System.out.printf(" %-65s %s%n", rel, formatSize(fi.getSize()));
+ }
+ if (files.size() > 15) {
+ System.out.printf(" ... 还有 %d 个文件%n", files.size() - 15);
+ }
+ }
+
+ /**
+ * 获取最后扫描的文件列表
+ */
+ public List getLastScanFiles() {
+ return lastScanFiles;
+ }
+}
diff --git a/connectors/connector-perf-test/src/main/java/io/tapdata/connector/paimon/perf/PerformanceTestRunner.java b/connectors/connector-perf-test/src/main/java/io/tapdata/connector/paimon/perf/PerformanceTestRunner.java
new file mode 100644
index 000000000..4c87e608a
--- /dev/null
+++ b/connectors/connector-perf-test/src/main/java/io/tapdata/connector/paimon/perf/PerformanceTestRunner.java
@@ -0,0 +1,1652 @@
+package io.tapdata.connector.paimon.perf;
+
+import io.tapdata.connector.paimon.config.PaimonConfig;
+import io.tapdata.connector.paimon.service.PaimonService;
+import io.tapdata.entity.event.dml.TapInsertRecordEvent;
+import io.tapdata.entity.event.dml.TapRecordEvent;
+import io.tapdata.entity.logger.Log;
+import io.tapdata.entity.schema.TapTable;
+import io.tapdata.entity.utils.DataMap;
+import io.tapdata.pdk.apis.context.TapConnectorContext;
+import io.tapdata.pdk.apis.spec.TapNodeSpecification;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.mockito.Mockito;
+
+import java.io.*;
+import java.text.SimpleDateFormat;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+/**
+ * Paimon 写入性能参数调优测试主类
+ *
+ * 运行方式：
+ *
+ * ./run-perf-test.sh [mode]
+ * mode: basic | buffer | target | bucket | compaction | nosmallfile | format | pkupdate | parallelism | all | auto
+ *
+ *
+ * 交互模式（默认）：每个用例前后按回车键继续；auto 模式：无需交互，全自动运行。
+ */
+public class PerformanceTestRunner {
+
+ // ─── 常量 ─────────────────────────────────────────────────────────────────
+
+ public static final String BASE_TEST_DIR = "/tmp/paimon-perf-test/";
+ private static final String DATABASE = "default";
+ private static final String TABLE_NAME = "test_table";
+ public static final int TOTAL_RECORDS = 5_000_000; // 数据集总大小
+ private static final int BATCH_SIZE = 100_000; // 每批次写入记录数，也是PaimonService 累积批次大小
+ private static final int INIT_TOTAL_RECORDS = 5_000_000; //模拟初始化阶段全表数据量
+
+ // ─── S3 测试配置 ──────────────────────────────────────────────────────────
+
+ /** 是否启用 S3 存储（false = 使用本地文件系统） */
+ private static final boolean ENABLE_S3 = true;
+
+ /** S3 端点地址 */
+ private static final String S3_ENDPOINT = "http://192.168.1.184:9080";
+// private static final String S3_ENDPOINT = "http://113.98.206.142:9080";
+
+ /** S3 访问密钥 */
+ private static final String S3_ACCESS_KEY = "admin";
+
+ /** S3 密钥 */
+ private static final String S3_SECRET_KEY = "admin123";
+
+ /** S3 区域（可选）:MinIO里 几乎没用，随便填一个合法字符串即可，不校验 region，不影响连接、不影响权限、不影响性能 */
+ private static final String S3_REGION = "us-east-1";
+
+ /** S3 仓库路径前缀（bucket 名称） */
+ public static final String S3_BUCKET = "s3://luke";
+ private static final String S3_WAREHOUSE = S3_BUCKET+"/warehouse-paimon-perf";
+
+ // ─── 实例变量 ─────────────────────────────────────────────────────────────
+
+ private final String baseDir;
+ private final String database;
+ private final String tableName;
+ private boolean interactive;
+ private final Log logger;
+
+ // ─── 构造函数 ─────────────────────────────────────────────────────────────
+
+ public PerformanceTestRunner(String baseDir, String database, String tableName) {
+ this.baseDir = baseDir;
+ this.database = database;
+ this.tableName = tableName;
+ this.interactive = true;
+ this.logger = buildConsoleLog();
+ }
+
+ public void setInteractive(boolean interactive) {
+ this.interactive = interactive;
+ }
+
+ // ─── 简单控制台 Log 实现 ──────────────────────────────────────────────────
+
+ private static Log buildConsoleLog() {
+ return new Log() {
+ @Override public void debug(String m, Object... p) { print("[DEBUG] ", m, p);}
+ @Override public void info(String m, Object... p) { print("[INFO] ", m, p); }
+ @Override public void warn(String m, Object... p) { print("[WARN] ", m, p); }
+ @Override public void error(String m, Object... p) { print("[ERROR]", m, p); }
+ @Override public void error(String m, Throwable t) { System.err.println("[ERROR] " + m + (t != null ? ": " + t.getMessage() : "")); }
+ @Override public void fatal(String m, Object... p) { print("[FATAL]", m, p); }
+ @Override public void trace(String m, Object... p) {}
+ private void print(String prefix, String m, Object[] p) {
+ if (p != null) { for (Object o : p) m = m.replaceFirst("\\{}", String.valueOf(o)); }
+ System.out.println(prefix + " " + m);
+ }
+ };
+ }
+
+ // ─── 测试用例目录管理 ────────────────────────────────────────────────────
+
+ private String warehouseForCase(TestCase tc) {
+ // 每个用例独立仓库，避免 schema 冲突
+ return baseDir + "/" + tc.getId();
+ }
+
+ // ─── PaimonService 工厂 ───────────────────────────────────────────────────
+
+ private PaimonService buildPaimonService(TestCase tc) throws Exception {
+ PaimonConfig config = new PaimonConfig();
+
+ // ── 核心参数 → PaimonConfig setters（会自动填充 tableProperties）──────────
+ applyConfigSetters(config, tc);
+ applyConfigSettersGlobal(config, tc);
+
+ PaimonService service = new PaimonService(config, logger);
+ service.init();
+ return service;
+ }
+
+ /**
+ * 全局参数生效：优先级高于TestCase中
+ *
+ * @param config
+ * @param tc
+ */
+ private void applyConfigSettersGlobal(PaimonConfig config, TestCase tc) {
+ // ── 存储类型和仓库路径 ──────────────────────────────────────────────
+ if (ENABLE_S3) {
+ // 使用 S3 存储
+ config.setStorageType("s3");
+ config.setWarehouse(S3_WAREHOUSE + "/" + tc.getId());
+ config.setS3Endpoint(S3_ENDPOINT);
+ config.setS3AccessKey(S3_ACCESS_KEY);
+ config.setS3SecretKey(S3_SECRET_KEY);
+ config.setS3Region(S3_REGION);
+ System.out.println(" >> 存储类型: S3 (" + S3_ENDPOINT + ")");
+ } else {
+ // 使用本地文件系统
+ config.setStorageType("local");
+ config.setWarehouse(warehouseForCase(tc));
+ System.out.println(" >> 存储类型: 本地文件系统，地址：" + config.getWarehouse());
+ }
+
+ config.setDatabase(database);
+// config.setBatchAccumulationSize(BATCH_SIZE);
+ //模拟全量+增量模式使用：
+ config.setCreateAutoInc(true);
+ config.setDiskTmpDir(BASE_TEST_DIR + "/tmp," + BASE_TEST_DIR + "/tmp2");
+ //为了验证Paimon参数的效果，测试中关闭：
+ config.setEnableAutoCompaction(false);
+
+ }
+
+ /**
+ * 将参数 Map 中的所有值自动写入到 PaimonConfig 的对应属性中
+ * 使用反射机制自动匹配参数名和 setter 方法，支持特殊映射和类型转换
+ *
+ * 支持的参数映射规则：
+ * 1. 特殊映射：通过 specialSetters 定义复杂映射关系（如单位转换、多属性设置等）
+ * 2. 反射自动映射：参数名转驼峰命名后匹配 setter 方法
+ * 3. 表属性降级：未匹配的参数放入 tableProperties，用于构建 Paimon 表选项
+ */
+ private void applyConfigSetters(PaimonConfig config, TestCase tc) {
+
+ // 批次累积大小
+ config.setBatchAccumulationSize(tc.getBatchSize());
+ Map params = tc.getParameters();
+
+ if (params == null || params.isEmpty()) return;
+
+ // ── 写入线程 / 并行度 ─────────────────────────────────────────────────
+ String parallelism = params.get("sink.parallelism");
+ if (parallelism != null) {
+ try { config.setWriteThreads(Integer.parseInt(parallelism)); }
+ catch (NumberFormatException ignored) {}
+ }
+
+ // 特殊参数映射表：参数名 → Setter 方法调用逻辑
+ Map specialSetters = new HashMap<>();
+
+ // 缓冲区大小（MB）
+ specialSetters.put("write-buffer-size", (cfg, val) -> cfg.setWriteBufferSize(parseSizeMb(val)));
+ // 目标文件大小（MB）
+ specialSetters.put("target-file-size", (cfg, val) -> cfg.setTargetFileSize(parseSizeMb(val)));
+ // 分桶策略
+ specialSetters.put("bucket", (cfg, val) -> {
+ int bucket = Integer.parseInt(val);
+ if (bucket > 0) {
+ cfg.setBucketMode("fixed");
+ cfg.setBucketCount(bucket);
+ } else if (bucket == -1 || bucket == -2) {
+ cfg.setBucketMode("dynamic");
+ cfg.setBucketCount(bucket);
+ }
+ });
+ // 写入线程/并行度
+ specialSetters.put("sink.parallelism", (cfg, val) -> cfg.setWriteThreads(Integer.parseInt(val)));
+ // 仅写入模式（禁用自动合并）:注释掉的原因为可以走下面的反射
+// specialSetters.put("enableAutoCompaction", (cfg, val) -> cfg.setEnableAutoCompaction(!Boolean.parseBoolean(val)));
+ // 缓冲区溢写
+ specialSetters.put("write-buffer-spillable", (cfg, val) -> cfg.setDiskOverflowWrite(Boolean.parseBoolean(val)));
+ // 溢写最大磁盘大小（GB）
+ specialSetters.put("write-buffer-spill.max-disk-size", (cfg, val) -> cfg.setDiskMaxSize(parseSizeGb(val)));
+ // 溢写临时目录
+ specialSetters.put("write-buffer-spill.tmp-dirs", (cfg, val) -> cfg.setDiskTmpDir(val));
+ // 提交间隔
+ specialSetters.put("commit-interval-ms", (cfg, val) -> cfg.setCommitIntervalMs(Integer.parseInt(val)));
+ // 异步提交
+ specialSetters.put("enable-async-commit", (cfg, val) -> cfg.setEnableAsyncCommit(Boolean.parseBoolean(val)));
+ // 合并间隔
+ specialSetters.put("compaction-interval-minutes", (cfg, val) -> cfg.setCompactionIntervalMinutes(Integer.parseInt(val)));
+ // 主键更新
+ specialSetters.put("enable-primary-key-update", (cfg, val) -> cfg.setEnablePrimaryKeyUpdate(Boolean.parseBoolean(val)));
+ // 分区键（逗号分隔）
+ specialSetters.put("partition-key", (cfg, val) -> cfg.setPartitionKey(Arrays.asList(val.split(","))));
+
+ for (Map.Entry entry : params.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ if (value == null || value.isEmpty()) continue;
+
+ try {
+ // 1. 优先检查特殊映射
+ ConfigSetter specialSetter = specialSetters.get(key);
+ if (specialSetter != null) {
+ specialSetter.set(config, value);
+ System.out.println(" [INFO] 参数='" + key + "'，值="+ value +"，通过Tapdata属性设置");
+ continue;
+ }
+
+ // 2. 尝试通过反射自动设置
+ if (applyConfigPropertyByReflection(config, key, value)) {
+ System.out.println(" [WARN] 参数='" + key + "' '，值="+ value +"，尝试通过反射自动设置成功");
+ continue;
+ }
+
+ // 3. 未匹配的参数放入 tableProperties（用于构建 Paimon 表选项）
+ put2TableProperties(config, key, value);
+ } catch (Exception e) {
+ System.err.println(" [ERROR] 设置配置属性失败: " + key + "=" + value + " - " + e.getMessage());
+ }
+ }
+ }
+
+ /**
+ * 通过反射自动调用 PaimonConfig 的 setter 方法
+ *
+ * @param config PaimonConfig 实例
+ * @param key 参数名（支持横线和下划线分隔，会自动转为驼峰）
+ * @param value 参数值（String 类型，会自动类型转换）
+ * @return true 如果成功找到并调用了对应的 setter 方法
+ */
+ private boolean applyConfigPropertyByReflection(PaimonConfig config, String key, String value) {
+ // 将参数名转为驼峰命名（例：write-buffer-size → writeBufferSize）
+ String camelKey = toCamelCase(key);
+
+ // 构造 setter 方法名
+ String setterName = "set" + Character.toUpperCase(camelKey.charAt(0)) + camelKey.substring(1);
+
+ try {
+ // 遍历所有可能的参数类型
+ Class[] paramTypes = {String.class, Integer.class, Boolean.class, List.class};
+
+ for (Class paramType : paramTypes) {
+ try {
+ java.lang.reflect.Method method = PaimonConfig.class.getMethod(setterName, paramType);
+ Object convertedValue = convertValue(value, paramType);
+ if (convertedValue != null) {
+ method.invoke(config, convertedValue);
+ return true;
+ }
+ } catch (NoSuchMethodException e) {
+ // 继续尝试下一个类型
+ }
+ }
+ } catch (Exception e) {
+ // 反射调用失败，返回 false 交由上层处理
+ }
+
+ return false;
+ }
+
+ /**
+ * 将横线/下划线分隔的字符串转为驼峰命名
+ * 例：write-buffer-size → writeBufferSize
+ * s3_endpoint → s3Endpoint
+ */
+ private String toCamelCase(String str) {
+ StringBuilder result = new StringBuilder();
+ boolean nextUpper = false;
+
+ for (int i = 0; i < str.length(); i++) {
+ char c = str.charAt(i);
+ if (c == '-' || c == '_' || c == '.') {
+ nextUpper = true;
+ } else {
+ if (nextUpper) {
+ result.append(Character.toUpperCase(c));
+ nextUpper = false;
+ } else {
+ result.append(c);
+ }
+ }
+ }
+
+ return result.toString();
+ }
+
+ /**
+ * 将 String 值转换为目标类型
+ */
+ private Object convertValue(String value, Class targetType) {
+ if (targetType == String.class) {
+ return value;
+ } else if (targetType == Integer.class) {
+ try {
+ return Integer.parseInt(value);
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ } else if (targetType == Boolean.class) {
+ return Boolean.parseBoolean(value);
+ } else if (targetType == List.class) {
+ return Arrays.asList(value.split(","));
+ }
+ return null;
+ }
+
+ /**
+ * 配置设置器函数式接口
+ */
+ @FunctionalInterface
+ private interface ConfigSetter {
+ void set(PaimonConfig config, String value);
+ }
+
+ /**
+ * 将未匹配的参数放入 PaimonConfig.tableProperties 中
+ * 这些属性将直接写入 Paimon 表的 OPTIONS 中，用于构建表配置
+ *
+ * @param config PaimonConfig 实例
+ * @param key 参数名
+ * @param value 参数值
+ */
+ private void put2TableProperties(PaimonConfig config, String key, String value) {
+ List> tableProperties = config.getTableProperties();
+
+ LinkedHashMap kv = new LinkedHashMap<>();
+ // 将参数添加到第一个 LinkedHashMap 中
+ kv.put("propKey", key);
+ kv.put("propValue", value);
+ tableProperties.add(kv);
+ System.out.println(" [INFO] 参数='" + key + "' '，值="+ value +"，通过paimon属性设置");
+ }
+
+ /**
+ * 解析带单位的大小，返回 MB 数（如 "512mb" → 512，"1gb" → 1024）
+ */
+ private static int parseSizeMb(String s) {
+ s = s.trim().toLowerCase();
+ if (s.endsWith("gb")) return (int) (Double.parseDouble(s.replace("gb", "").trim()) * 1024);
+ if (s.endsWith("mb")) return Integer.parseInt(s.replace("mb", "").trim());
+ if (s.endsWith("kb")) return (int) (Double.parseDouble(s.replace("kb", "").trim()) / 1024);
+ return Integer.parseInt(s);
+ }
+
+ /**
+ * 解析带单位的大小，返回 GB 数
+ */
+ private static int parseSizeGb(String s) {
+ s = s.trim().toLowerCase();
+ if (s.endsWith("tb")) return (int) (Double.parseDouble(s.replace("tb", "").trim()) * 1024);
+ if (s.endsWith("gb")) return Integer.parseInt(s.replace("gb", "").trim());
+ if (s.endsWith("mb")) return (int) (Double.parseDouble(s.replace("mb", "").trim()) / 1024);
+ return Integer.parseInt(s);
+ }
+
+ // ─── 创建测试表 ────────────────────────────────────────────────────────────
+
+ private TapTable createTapTable() {
+ DataGenerator dg = new DataGenerator(0, tableName);
+ return dg.generateTapTable();
+ }
+
+ private void createFreshTable(PaimonService service) throws Exception {
+ try { service.dropTable(tableName); } catch (Exception ignored) {}
+ TapTable table = createTapTable();
+ service.createTable(table);
+ }
+
+ // ─── 核心执行方法 ──────────────────────────────────────────────────────────
+
+ /**
+ * 执行单个测试用例
+ */
+ public TestResult runTestCase(TestCase tc) {
+ printSeparator("=");
+ System.out.printf(" 用例 %-8s: %s%n", tc.getId(), tc.getName());
+ System.out.printf(" 组 别: %-20s 描述: %s%n", tc.getGroup(), tc.getDescription());
+ printSeparator("-");
+ printParameters(tc.getParameters());
+
+ if (interactive) {
+ System.out.println("\n [按 Enter 开始本用例，Ctrl+C 退出]");
+ waitForEnter();
+ }
+
+ PaimonService service = null;
+ long startMs = 0;
+ long endMs = 0;
+ AtomicLong written = new AtomicLong(0);
+ String error = null;
+ PaimonFileObserver observer;
+ if (ENABLE_S3) {
+ // S3 模式的观测器
+ observer = new PaimonFileObserver(
+ S3_WAREHOUSE + "/" + tc.getId(),
+ database,
+ tableName,
+ S3_ENDPOINT,
+ S3_ACCESS_KEY,
+ S3_SECRET_KEY,
+ S3_REGION
+ );
+ } else {
+ // 本地模式的观测器
+ observer = new PaimonFileObserver(warehouseForCase(tc), database, tableName);
+ }
+
+ try {
+ // 1. 初始化服务并创建表
+ System.out.println("\n >> 初始化 PaimonService...");
+ service = buildPaimonService(tc);
+ createFreshTable(service);
+
+ // 2. 验证表参数是否生效
+ System.out.println(" >> 验证表配置参数...");
+ validateTableParameters(tc, service);
+
+ TapConnectorContext tapConnectorContext = new TapConnectorContext(Mockito.mock(TapNodeSpecification.class), new DataMap(), new DataMap(), logger);
+
+ System.out.printf(" >> 仓库路径: %s%n", warehouseForCase(tc));
+ System.out.printf(" >> 开始写入 %,d 条记录 (主键重复率 %d%%, QPS限制 %s)%n",
+ tc.getDataSize(), tc.getPrimaryKeyDuplicateRate(),
+ tc.getQps() > 0 ? tc.getQps() + "" : "无限制");
+
+ // 显示用例级别的参数覆盖
+ if (tc.getBatchSize() != null || tc.getInitTotalRecords() != null) {
+ System.out.println(" >> 用例级别参数覆盖:");
+ if (tc.getBatchSize() != null) {
+ System.out.printf(" - batchSize = %,d (全局: %,d)%n", tc.getBatchSize(), BATCH_SIZE);
+ }
+ if (tc.getInitTotalRecords() != null) {
+ System.out.printf(" - initTotalRecords = %,d (全局: %,d)%n", tc.getInitTotalRecords(), INIT_TOTAL_RECORDS);
+ }
+ }
+
+ // 2. 执行写入
+ TapTable tapTable = createTapTable();
+ DataGenerator gen = new DataGenerator(tc.getPrimaryKeyDuplicateRate(), tableName);
+
+ // 使用 TestCase 独立的参数（如果设置了），否则使用全局常量
+ long totalRecordsToWrite = tc.getDataSize();
+ Integer caseBatchSize = tc.getBatchSize();
+ int effectiveBatchSize = caseBatchSize != null ? caseBatchSize : BATCH_SIZE;
+ Integer caseInitTotal = tc.getInitTotalRecords();
+ int effectiveInitTotal = caseInitTotal != null ? caseInitTotal : INIT_TOTAL_RECORDS;
+
+ startMs = System.currentTimeMillis();
+ long total = totalRecordsToWrite;
+ long remain = total;
+ long qpsSlotStartMs = System.currentTimeMillis();
+ long qpsSlotWritten = 0;
+
+ while (remain > 0) {
+ int batchSz = (int) Math.min(effectiveBatchSize, remain);
+ List batch = new ArrayList<>(batchSz);
+ for (int i = 0; i < batchSz; i++) {
+ Map rec = gen.generateRecord();
+ TapInsertRecordEvent evt = new TapInsertRecordEvent();
+ evt.setAfter(rec);
+ evt.setTableId(tableName);
+ evt.setReferenceTime(System.currentTimeMillis());
+ Map info = new HashMap<>(1);
+ info.put("batchOffset",i);
+ evt.setInfo(info);
+ batch.add(evt);
+ if (qpsSlotWritten + effectiveBatchSize >= effectiveInitTotal) {
+ // 模拟 已写数据量 大于 初始化阶段全表数据量 为增量cdc：
+ evt.getInfo().put(TapRecordEvent.INFO_KEY_SYNC_STAGE, "CDC");
+ }
+ }
+ service.writeRecords(batch, tapTable, tapConnectorContext);
+ remain -= batchSz;
+ written.addAndGet(batchSz);
+ qpsSlotWritten += batchSz;
+
+ // QPS throttle
+ if (tc.getQps() > 0) {
+ long elapsed = System.currentTimeMillis() - qpsSlotStartMs;
+ long expected = qpsSlotWritten * 1000L / tc.getQps();
+ long sleep = expected - elapsed;
+ if (sleep > 0) {
+ try { Thread.sleep(sleep); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
+ }
+ }
+
+ // 进度打印
+ long pct = (written.get() * 100) / total;
+ if (written.get() % (effectiveBatchSize * 10) == 0 || remain == 0) {
+ double elapsed = (System.currentTimeMillis() - startMs) / 1000.0;
+ double throughput = elapsed > 0 ? written.get() / elapsed : 0;
+ System.out.printf(" >> 进度: %,d/%,d (%d%%) | 吞吐: %.0f 条/秒%n",
+ written.get(), total, pct, throughput);
+ }
+ }
+
+ // 3. 强制 flush
+ System.out.println(" >> 执行最终 flush...");
+ service.flushAll();
+ endMs = System.currentTimeMillis();
+
+ } catch (Throwable e) {
+ endMs = System.currentTimeMillis();
+ error = e.getClass().getSimpleName() + ": " + e.getMessage();
+ System.err.println(" [ERROR] 用例执行异常: " + error);
+ printStackTrace(e);
+ } finally {
+ if (service != null) {
+ try { service.close(); } catch (Exception ignored) {}
+ }
+ }
+
+ // 4. 统计文件
+ List files = Collections.emptyList();
+ try { files = observer.scanAllFiles(); } catch (Exception e) {
+ System.err.println(" [WARN] 文件扫描失败: " + e.getMessage());
+ }
+
+ long durationMs = endMs - startMs;
+ double throughput = durationMs > 0 ? written.get() * 1000.0 / durationMs : 0;
+
+ TestResult result = new TestResult(tc, written.get(), durationMs, throughput, files, error);
+
+ // 5. 打印结果
+ printResult(result, observer);
+
+ if (interactive) {
+ System.out.println("\n [按 Enter 继续下一个用例]");
+ waitForEnter();
+ }
+ return result;
+ }
+
+ public static void printStackTrace(Throwable e) {
+ e.printStackTrace();
+ }
+
+ /**
+ * 运行整个测试组
+ */
+ public List runTestGroup(String groupName) throws Exception {
+ List cases;
+ switch (groupName.toLowerCase()) {
+ case "basic": cases = TestCase.createBasicTests(); break;
+ case "buffer": cases = TestCase.createWriteBufferTests(); break;
+ case "target": cases = TestCase.createTargetFileSizeTests(); break;
+ case "bucket": cases = TestCase.createBucketTests(); break;
+ case "compaction": cases = TestCase.createCompactionTests(); break;
+ case "nosmallfile": cases = TestCase.createNoSmallFileTests(); break;
+ case "format": cases = TestCase.createFormatCompressionTests(); break;
+ case "pkupdate": cases = TestCase.createPrimaryKeyUpdateTests(); break;
+ case "parallelism": cases = TestCase.createParallelismTests(); break;
+ case "all": cases = TestCase.createAllTests(); break;
+ default:
+ System.out.println(" [WARN] 未知测试组: " + groupName);
+ System.out.println(" 可用: " + String.join(", ", TestCase.getGroupDescriptions().keySet()));
+ return Collections.emptyList();
+ }
+
+ printSeparator("=");
+ System.out.printf(" 运行测试组: %s 共 %d 个用例%n", groupName, cases.size());
+ printSeparator("=");
+
+ List results = new ArrayList<>();
+ for (int i = 0; i < cases.size(); i++) {
+ TestCase tc = cases.get(i);
+ System.out.printf("%n >>> 用例 [%d/%d]%n", i + 1, cases.size());
+ results.add(runTestCase(tc));
+ }
+ return results;
+ }
+
+ // ─── 打印工具 ──────────────────────────────────────────────────────────────
+
+ private static void printSeparator(String ch) {
+ System.out.println(StringUtils.repeat(ch, 70));
+ }
+
+ /**
+ * 从 Paimon Catalog 中读取表的实际配置参数（支持本地和 S3）
+ */
+ private Map readActualTableOptions(String warehouse, String database, String tableName) {
+ try {
+ Options catalogOptions = new Options();
+
+ if (ENABLE_S3) {
+ // S3 模式：使用 s3:// 协议（Paimon 原生 S3 FileIO）而非 s3a://
+ catalogOptions.set("warehouse", warehouse);
+
+ // Paimon S3 配置
+ catalogOptions.set("s3.endpoint", S3_ENDPOINT);
+ catalogOptions.set("s3.access-key", S3_ACCESS_KEY);
+ catalogOptions.set("s3.secret-key", S3_SECRET_KEY);
+
+ if (S3_REGION != null && !S3_REGION.isEmpty()) {
+ catalogOptions.set("s3.region", S3_REGION);
+ }
+
+ // 路径样式访问（MinIO 需要）
+ catalogOptions.set("s3.path-style-access", "true");
+
+ // 禁用 SSL（如果是 http 端点）
+ if (S3_ENDPOINT.startsWith("http://")) {
+ catalogOptions.set("s3.ssl.enabled", "false");
+ }
+ } else {
+ // 本地模式
+ catalogOptions.set("warehouse", warehouse);
+ }
+
+ CatalogContext context = CatalogContext.create(catalogOptions);
+ Catalog catalog = CatalogFactory.createCatalog(context);
+ Identifier identifier = Identifier.create(database, tableName);
+ Table table = catalog.getTable(identifier);
+ Map options = table.options();
+ catalog.close();
+ return options;
+ } catch (org.apache.paimon.fs.UnsupportedSchemeException e) {
+ // S3 协议不可用，可能是缺少 paimon-s3 依赖
+ System.err.println(" [WARN] S3 文件系统不可用: " + e.getMessage());
+ System.err.println(" [提示] 请确保项目中包含 paimon-s3 依赖");
+ return Collections.emptyMap();
+ } catch (Exception e) {
+ System.err.println(" [WARN] 读取表配置失败: " + e.getMessage());
+ if (System.getProperty("perf.verbose", "false").equals("true")) {
+ e.printStackTrace();
+ }
+ return Collections.emptyMap();
+ }
+ }
+
+ /**
+ * 验证表参数是否已生效：对比预期参数和实际表配置（支持本地和 S3）
+ */
+ private void validateTableParameters(TestCase tc, PaimonService service) {
+ Map expectedParams = tc.getParameters();
+
+ // 根据存储类型构建仓库路径
+ String warehousePath;
+ if (ENABLE_S3) {
+ warehousePath = S3_WAREHOUSE + "/" + tc.getId();
+ System.out.println(" >> 验证存储: S3 (" + S3_ENDPOINT + ")");
+ } else {
+ warehousePath = warehouseForCase(tc);
+ System.out.println(" >> 验证存储: 本地文件系统");
+ }
+
+ Map actualOptions = readActualTableOptions(warehousePath, database, tableName);
+
+ if (actualOptions.isEmpty()) {
+ System.out.println(" [WARN] 无法读取表配置，跳过参数验证");
+ return;
+ }
+
+ // 分类验证
+ List serviceOnlyParams = Arrays.asList(
+ "write-buffer-spillable", "write-buffer-spill.max-disk-size"
+ );
+
+ System.out.println(" 参数验证结果:");
+ int validated = 0;
+ int matched = 0;
+ int mismatched = 0;
+
+ for (Map.Entry entry : expectedParams.entrySet()) {
+ String key = entry.getKey();
+ String expectedValue = entry.getValue();
+
+ // Service 专有参数不在表选项中
+ if (serviceOnlyParams.contains(key)) {
+ System.out.printf(" [Service] %-45s = %-20s ✓ 作用于 PaimonService%n", key, expectedValue);
+ validated++;
+ continue;
+ }
+
+ // 检查表选项中是否有该参数
+ String actualValue = null;
+ for (Map.Entry
+
+
+
+
+
+
+
+
+
+
+
+
+
+