[core] Align primary-key index maintenance with data levels

docs/docs/primary-key-table/global-index.mdx

@@ -234,8 +234,6 @@ schema validation.
 | `fields.<column>.pk-btree.index.options` | Not set | JSON object containing BTree build options. Unqualified keys are scoped to `btree-index`. |
 | `pk-bitmap.index.columns` | Not set | Comma-separated columns which own independent Bitmap indexes. |
 | `fields.<column>.pk-bitmap.index.options` | Not set | JSON object containing Bitmap build options. Unqualified keys are scoped to `bitmap-index`. |
-| `fields.<column>.pk-index.compaction.level-fanout` | `5` | Number of similarly sized index groups which triggers a rebuild and maximum row-count ratio within one size tier. Shared by all four families. Must be greater than `1`. |
-| `fields.<column>.pk-index.compaction.stale-ratio-threshold` | `0.2` | Ratio of rows from inactive source files which triggers a rebuild. Shared by all four families. Must be in `(0, 1]`. |
 | `global-index.search-mode` | `fast` | Search mode for primary-key Vector and Full Text queries. `fast` searches indexed data only, so uncovered files are omitted. For Vector, `full` and `detail` search uncovered files exactly. Primary-key Full Text supports only `fast`. |
 
 For algorithm-specific options, see the corresponding
@@ -259,17 +257,16 @@ inside real buckets. These rows become visible after batch compaction publishes
 buckets. Indexes are created when that process physically rewrites the rows into eligible compact
 output; simply assigning or upgrading a pending file does not make it an index source.
 
-### Index LSM Maintenance
+### Data-Level Maintenance
 
-Each indexed column maintains its own immutable index groups. Maintenance uses the shared
-field-scoped compaction options:
+Each indexed column maintains one immutable index payload for the complete eligible source-file
+set in every non-zero data level. When data compaction changes a level, Paimon rebuilds that whole
+level payload, including files in the target level which were not direct compaction inputs. A
+level payload is used only when its ordered source names and row counts exactly match the current
+data level; partial, duplicate, stale, and cross-level payloads are rejected.
 
-- When at least `level-fanout` similarly sized groups exist, Paimon rebuilds them into a larger
- group. The largest selected group can contain at most `level-fanout` times the rows of the
- smallest group.
-- When the ratio of rows belonging to inactive source files reaches
- `stale-ratio-threshold`, Paimon rebuilds the affected group from its remaining active sources.
-- A rebuild atomically replaces its input groups after the new group is complete.
+A rebuild atomically replaces the old payload after the complete new payload is ready. Unrelated
+data levels retain their existing payloads.
 
 Index construction can execute asynchronously inside the writer. A writer which waits for
 compaction also waits for active index maintenance; a non-blocking writer can complete maintenance

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -4299,22 +4299,6 @@ public boolean primaryKeyFullTextIndexEnabled() {
  return options.getOptional(PK_FULL_TEXT_INDEX_COLUMNS).isPresent();
  }
 
- public int primaryKeyIndexCompactionLevelFanout(String column) {
- return options.getInteger(primaryKeyIndexCompactionLevelFanoutKey(column), 5);
- }
-
- public double primaryKeyIndexCompactionStaleRatioThreshold(String column) {
- return options.getDouble(primaryKeyIndexCompactionStaleRatioThresholdKey(column), 0.2);
- }
-
- public static String primaryKeyIndexCompactionLevelFanoutKey(String column) {
- return "fields." + column + ".pk-index.compaction.level-fanout";
- }
-
- public static String primaryKeyIndexCompactionStaleRatioThresholdKey(String column) {
- return "fields." + column + ".pk-index.compaction.stale-ratio-threshold";
- }
-
  public List<String> primaryKeyVectorIndexColumns() {
  return primaryKeyIndexColumns(PK_VECTOR_INDEX_COLUMNS);
  }

paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java

@@ -381,21 +381,15 @@ public static Factory create(
  readerFactoryBuilder,
  field,
  definition.indexType(),
- definition.options(),
- definition.compactionLevelFanout(),
- definition.compactionStaleRatioThreshold()));
+ definition.options()));
  break;
  case FULL_TEXT:
  checkArgument(
  fullTextFactory == null,
  "Only one primary-key full-text index is supported.");
  fullTextFactory =
  new FullTextDefinitionFactory(
- readerFactoryBuilder,
- field,
- definition.options(),
- definition.compactionLevelFanout(),
- definition.compactionStaleRatioThreshold());
+ readerFactoryBuilder, field, definition.options());
  break;
  default:
  throw new IllegalArgumentException(
@@ -480,20 +474,14 @@ private static final class FullTextDefinitionFactory {
  private final KeyValueFileReaderFactory.Builder readerFactoryBuilder;
  private final DataField field;
  private final org.apache.paimon.options.Options options;
- private final int compactionLevelFanout;
- private final double compactionStaleRatioThreshold;
 
  private FullTextDefinitionFactory(
  KeyValueFileReaderFactory.Builder readerFactoryBuilder,
  DataField field,
- org.apache.paimon.options.Options options,
- int compactionLevelFanout,
- double compactionStaleRatioThreshold) {
+ org.apache.paimon.options.Options options) {
  this.readerFactoryBuilder = readerFactoryBuilder;
  this.field = field;
  this.options = options;
- this.compactionLevelFanout = compactionLevelFanout;
- this.compactionStaleRatioThreshold = compactionStaleRatioThreshold;
  }
 
  private BucketedFullTextIndexMaintainer create(
@@ -515,8 +503,6 @@ private BucketedFullTextIndexMaintainer create(
  field.id(),
  indexFile,
  builder,
- compactionLevelFanout,
- compactionStaleRatioThreshold,
  restoredDataFiles,
  restoredPayloads,
  executor);
@@ -529,22 +515,16 @@ private static final class SortedDefinitionFactory {
  private final DataField field;
  private final String indexType;
  private final org.apache.paimon.options.Options options;
- private final int compactionLevelFanout;
- private final double compactionStaleRatioThreshold;
 
  private SortedDefinitionFactory(
  KeyValueFileReaderFactory.Builder readerFactoryBuilder,
  DataField field,
  String indexType,
- org.apache.paimon.options.Options options,
- int compactionLevelFanout,
- double compactionStaleRatioThreshold) {
+ org.apache.paimon.options.Options options) {
  this.readerFactoryBuilder = readerFactoryBuilder;
  this.field = field;
  this.indexType = indexType;
  this.options = options;
- this.compactionLevelFanout = compactionLevelFanout;
- this.compactionStaleRatioThreshold = compactionStaleRatioThreshold;
  }
 
  private BucketedSortedIndexMaintainer create(
@@ -570,8 +550,6 @@ private BucketedSortedIndexMaintainer create(
  indexType,
  indexFile,
  builder::build,
- compactionLevelFanout,
- compactionStaleRatioThreshold,
  restoredDataFiles,
  restoredPayloads,
  executor);

paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java

@@ -36,24 +36,14 @@ public enum Family {
  private final String indexType;
  private final Options options;
  private final Family family;
- private final int compactionLevelFanout;
- private final double compactionStaleRatioThreshold;
 
  public PrimaryKeyIndexDefinition(
- String column,
- int fieldId,
- String indexType,
- Options options,
- Family family,
- int compactionLevelFanout,
- double compactionStaleRatioThreshold) {
+ String column, int fieldId, String indexType, Options options, Family family) {
  this.column = column;
  this.fieldId = fieldId;
  this.indexType = indexType;
  this.options = options;
  this.family = family;
- this.compactionLevelFanout = compactionLevelFanout;
- this.compactionStaleRatioThreshold = compactionStaleRatioThreshold;
  }
 
  public String column() {
@@ -75,12 +65,4 @@ public Options options() {
  public Family family() {
  return family;
  }
-
- public int compactionLevelFanout() {
- return compactionLevelFanout;
- }
-
- public double compactionStaleRatioThreshold() {
- return compactionStaleRatioThreshold;
- }
 }

paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java

@@ -63,39 +63,31 @@ public static PrimaryKeyIndexDefinitions create(TableSchema schema) {
  field.id(),
  BTreeGlobalIndexerFactory.IDENTIFIER,
  options.primaryKeyBTreeIndexOptions(column),
- PrimaryKeyIndexDefinition.Family.BTREE,
- options.primaryKeyIndexCompactionLevelFanout(column),
- options.primaryKeyIndexCompactionStaleRatioThreshold(column)));
+ PrimaryKeyIndexDefinition.Family.BTREE));
  } else if (bitmapColumns.contains(column)) {
  definitions.add(
  new PrimaryKeyIndexDefinition(
  column,
  field.id(),
  BitmapGlobalIndexerFactory.IDENTIFIER,
  options.primaryKeyBitmapIndexOptions(column),
- PrimaryKeyIndexDefinition.Family.BITMAP,
- options.primaryKeyIndexCompactionLevelFanout(column),
- options.primaryKeyIndexCompactionStaleRatioThreshold(column)));
+ PrimaryKeyIndexDefinition.Family.BITMAP));
  } else if (vectorColumns.contains(column)) {
  definitions.add(
  new PrimaryKeyIndexDefinition(
  column,
  field.id(),
  options.primaryKeyVectorIndexType(column),
  options.primaryKeyVectorIndexOptions(column),
- PrimaryKeyIndexDefinition.Family.VECTOR,
- options.primaryKeyIndexCompactionLevelFanout(column),
- options.primaryKeyIndexCompactionStaleRatioThreshold(column)));
+ PrimaryKeyIndexDefinition.Family.VECTOR));
  } else if (fullTextColumns.contains(column)) {
  definitions.add(
  new PrimaryKeyIndexDefinition(
  column,
  field.id(),
  "full-text",
  options.primaryKeyFullTextIndexOptions(column),
- PrimaryKeyIndexDefinition.Family.FULL_TEXT,
- options.primaryKeyIndexCompactionLevelFanout(column),
- options.primaryKeyIndexCompactionStaleRatioThreshold(column)));
+ PrimaryKeyIndexDefinition.Family.FULL_TEXT));
  }
  }