// whether to enable delta join
// AUTO: Optimizer will try to use delta join first. If failed, it will fallback to regular join.
// FORCE: Enforce to use delta join. If failed, an exception will be thrown.
// NONE: Don't try to use delta join.
// default is 'AUTO'
table.optimizer.delta-join.strategy: AUTO/FORCE/NONE
 
// whether to enable the cache in delta join
// default is true
table.exec.delta-join.cache-enable: true/false
 
// the cache size of the left side in delta join
// default is 10000L
table.exec.delta-join.left-cache-size: 10000L
 
// the cache size of the right side in delta join
// default is 10000L
table.exec.delta-join.right-cache-size: 10000L

/**
 * An index is a copy of selected columns of data, from a table, that is designed to enable very
 * efficient search. An index normally includes a "key" or direct link to the original row of data
 * from which it was copied, to allow the complete row to be retrieved efficiently.
 */
@Experimental
public interface Index {

    /** Get the name of this index. */
    String getName();

    /** Get the columns of this index. */
    List<String> getColumns();

    /** Print the index in a readable way. */
    String asSummaryString();
}

@PublicEvolving
  public final class Schema {

    ...
    private final List<UnresolvedIndex> indexes;

    public Schema(
      List<UnresolvedColumn> columns,
      List<UnresolvedWatermarkSpec> watermarkSpecs,
      @Nullable UnresolvedPrimaryKey primaryKey) {
      this(columns, watermarkSpecs, primaryKey, Collections.emptyList());
    }

    public Schema(
      List<UnresolvedColumn> columns,
      List<UnresolvedWatermarkSpec> watermarkSpecs,
      @Nullable UnresolvedPrimaryKey primaryKey,
      List<UnresolvedIndex> indexes) {
      ...
    }

    ...

    public List<UnresolvedIndex> getIndexes() {
      return indexes;
    }

    ...

    @PublicEvolving
      public static final class Builder {

        ...

        private final List<UnresolvedIndex> indexes;

        ...
      }

    /**
         * Declares an index for a set of given columns. Indexes are designed to enable very
         * efficient search. The indexes are informational only and can be used for optimizations.
         * It is the data owner's responsibility to guarantee the index queries allow the complete
         * row to be retrieved efficiently.
         *
         * <p>The index will be assigned a generated name in the format {@code INDEX_col1_col2}.
         *
         * @param columns indexes that form a table index
         */
    public Builder index(String... columns) {
      Preconditions.checkNotNull(indexes, "Index column names must not be null.");
      return index(Arrays.asList(columns));
    }

    /**
         * Declares an index for a set of given columns. Indexes are designed to enable very
         * efficient search. The indexes are informational only and can be used for optimizations.
         * It is the data owner's responsibility to guarantee the index queries allow the complete
         * row to be retrieved efficiently.
         *
         * <p>The index will be assigned a generated name in the format {@code INDEX_col1_col2}.
         *
         * @param columns indexes that form a table index
         */
    public Builder index(List<String> columns) {
      Preconditions.checkNotNull(indexes, "Index column names must not be null.");
      final String generatedIndexName =
        columns.stream().collect(Collectors.joining("_", "INDEX_", ""));
      return indexNamed(generatedIndexName, columns);
    }

    /**
         * Declares an index for a set of given columns. Indexes are designed to enable very
         * efficient search. The indexes are informational only and can be used for optimizations.
         * It is the data owner's responsibility to guarantee the index queries allow the complete
         * row to be retrieved efficiently.
         *
         * @param indexName the name of the index
         * @param columns columns that form a table index
         */
    public Builder indexNamed(String indexName, List<String> columns) {
      Preconditions.checkNotNull(indexName, "Index name must not be null.");
      Preconditions.checkNotNull(columns, "Index column names must not be null.");
      this.indexes.add(new UnresolvedIndex(indexName, columns));
      return this;
    }

    /** Declaration of an index that will be resolved to {@link Index} during schema resolution. */
    @Experimental
      public static final class UnresolvedIndex {
        private final String indexName;
        private final List<String> columnNames;

        public UnresolvedIndex(String indexName, List<String> columnNames) {
          this.indexName = indexName;
          this.columnNames = columnNames;
        }

        // getters / equals / hashCode
        ...

        @Override
        public String toString() {
          return String.format(
            "INDEX %s (%s)",
            EncodingUtils.escapeIdentifier(getIndexName()),
            getColumnNames().stream()
            .map(EncodingUtils::escapeIdentifier)
            .collect(Collectors.joining(", ")));
        }
      }
  }

@PublicEvolving
public final class ResolvedSchema {

    ...
    private final List<Index> indexes;

    public ResolvedSchema(
            List<Column> columns,
            List<WatermarkSpec> watermarkSpecs,
            @Nullable UniqueConstraint primaryKey) {
        this(columns, watermarkSpecs, primaryKey, Collections.emptyList());
    }

    public ResolvedSchema(
            List<Column> columns,
            List<WatermarkSpec> watermarkSpecs,
            @Nullable UniqueConstraint primaryKey,
            List<Index> indexes) {
        ...
    }

    // Getter
    ...
}