Core, Spark: SCALAR Key Lookup Index — reference implementation of #16961

api/src/main/java/org/apache/iceberg/index/IndexCatalog.java

@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.index;
+
+import java.util.List;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+
+/**
+ * Catalog interface for managing secondary indexes on Iceberg tables.
+ *
+ * <p>An index catalog stores a pointer to the current index metadata file for each registered
+ * index. Commits are atomic: {@link #updateIndex} replaces the metadata pointer only if the current
+ * pointer still matches the base that the caller read.
+ */
+public interface IndexCatalog {
+
+  /**
+   * Register a new index in the catalog.
+   *
+   * @param identifier the index identifier
+   * @param metadata the initial index metadata (must have a metadataFileLocation set)
+   * @throws AlreadyExistsException if an index with the same identifier already exists
+   */
+  void createIndex(IndexIdentifier identifier, IndexMetadata metadata);
+
+  /**
+   * Load the current metadata for an index.
+   *
+   * @param identifier the index identifier
+   * @return the current index metadata
+   * @throws NoSuchTableException if the index does not exist
+   */
+  IndexMetadata loadIndex(IndexIdentifier identifier);
+
+  /**
+   * Atomically replace the current index metadata.
+   *
+   * <p>The update succeeds only if the current metadata file location matches {@code
+   * base.metadataFileLocation()}. If another writer has already committed a newer version, this
+   * call throws {@link java.util.ConcurrentModificationException}.
+   *
+   * @param identifier the index identifier
+   * @param base the metadata the caller read (used for optimistic concurrency check)
+   * @param updated the new metadata to commit (must have a new metadataFileLocation)
+   */
+  void updateIndex(IndexIdentifier identifier, IndexMetadata base, IndexMetadata updated);
+
+  /**
+   * Remove an index from the catalog.
+   *
+   * <p>Does not delete the underlying index files — callers are responsible for cleanup.
+   *
+   * @param identifier the index identifier
+   * @throws NoSuchTableException if the index does not exist
+   */
+  void dropIndex(IndexIdentifier identifier);
+
+  /**
+   * Check if an index exists.
+   *
+   * @param identifier the index identifier
+   * @return true if the index exists
+   */
+  boolean indexExists(IndexIdentifier identifier);
+
+  /**
+   * List all indexes registered for a table.
+   *
+   * @param tableIdentifier the table identifier
+   * @return list of index metadata for all indexes on the table, may be empty
+   */
+  List<IndexMetadata> listIndexes(TableIdentifier tableIdentifier);
+}

api/src/main/java/org/apache/iceberg/index/IndexIdentifier.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.index;
+
+import java.util.Objects;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * Identifies a secondary index by its source table and index name.
+ *
+ * <p>Format: {@code namespace.table.indexName}
+ *
+ * <p>Example: {@code db.orders.order_id_idx}
+ */
+public class IndexIdentifier {
+
+  private final TableIdentifier tableIdentifier;
+  private final String name;
+
+  private IndexIdentifier(TableIdentifier tableIdentifier, String name) {
+    this.tableIdentifier = tableIdentifier;
+    this.name = name;
+  }
+
+  public static IndexIdentifier of(TableIdentifier tableIdentifier, String name) {
+    Preconditions.checkNotNull(tableIdentifier, "tableIdentifier is required");
+    Preconditions.checkArgument(name != null && !name.isEmpty(), "index name must be non-empty");
+    return new IndexIdentifier(tableIdentifier, name);
+  }
+
+  public TableIdentifier tableIdentifier() {
+    return tableIdentifier;
+  }
+
+  public String name() {
+    return name;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof IndexIdentifier)) {
+      return false;
+    }
+    IndexIdentifier that = (IndexIdentifier) o;
+    return Objects.equals(tableIdentifier, that.tableIdentifier) && Objects.equals(name, that.name);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(tableIdentifier, name);
+  }
+
+  @Override
+  public String toString() {
+    return tableIdentifier + "." + name;
+  }
+}

api/src/main/java/org/apache/iceberg/index/IndexMetadata.java

@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.index;
+
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+/**
+ * Metadata for a secondary index on an Iceberg table.
+ *
+ * <p>Index metadata is stored as a JSON file in the index's {@link #location()}. It contains the
+ * index definition (type, transform, key columns) and a list of {@link IndexSnapshot index
+ * snapshots}, each corresponding to a specific source table snapshot.
+ *
+ * <p>This interface matches the index metadata format defined in the Iceberg secondary index
+ * specification (format/index.md).
+ */
+public interface IndexMetadata {
+
+  int SUPPORTED_INDEX_FORMAT_VERSION = 1;
+  int DEFAULT_INDEX_FORMAT_VERSION = 1;
+
+  /** Index metadata format version. Currently 1. */
+  int formatVersion();
+
+  /** Stable UUID assigned when the index is created. */
+  String uuid();
+
+  /** UUID of the source table this index is built on. */
+  String tableUuid();
+
+  /** Base location used to create index file paths (metadata, tracking, leaf files). */
+  String location();
+
+  /**
+   * Logical index type, e.g. {@code "SCALAR"}, {@code "IVF"}, {@code "TERM"}.
+   *
+   * <p>Engines that do not recognize the type must skip this index and fall back to normal
+   * planning.
+   */
+  String type();
+
+  /**
+   * Transform function applied to key columns, e.g. {@code "HASH"} or {@code "IDENTITY"}.
+   *
+   * <p>Determines how index entries are organized within leaf files.
+   */
+  String transformFunction();
+
+  /**
+   * Source-table column IDs that the transform is applied to (key columns).
+   *
+   * <p>Leaf files are organized by the transform value derived from these columns.
+   */
+  List<Integer> keyColumnIds();
+
+  /**
+   * Optional source-table column IDs copied into the index for read convenience (included columns).
+   *
+   * <p>These columns are stored in leaf files but do not affect index organization. Used to serve
+   * covering queries without reading the source table.
+   */
+  List<Integer> includedColumnIds();
+
+  /**
+   * Optional index-level properties, e.g. {@code hash.num-buckets}, {@code ivf.distance-function}.
+   *
+   * @return an unmodifiable map of string properties, never null
+   */
+  Map<String, String> properties();
+
+  /**
+   * ID of the current index snapshot, or null if no snapshot has been committed yet.
+   *
+   * @return the current snapshot ID, or null
+   */
+  @Nullable
+  Long currentSnapshotId();
+
+  /** All index snapshots for this index, ordered by snapshot ID. */
+  List<IndexSnapshot> snapshots();
+
+  /**
+   * Location of the index metadata file, or null if not persisted yet.
+   *
+   * @return the metadata file location, or null
+   */
+  @Nullable
+  String metadataFileLocation();
+
+  // ---------------------------------------------------------------------------
+  // Derived helpers
+  // ---------------------------------------------------------------------------
+
+  /** Returns the current {@link IndexSnapshot}, or null if no snapshot has been committed. */
+  default IndexSnapshot currentSnapshot() {
+    Long id = currentSnapshotId();
+    if (id == null) {
+      return null;
+    }
+    for (IndexSnapshot snap : snapshots()) {
+      if (snap.snapshotId() == id) {
+        return snap;
+      }
+    }
+    return null;
+  }
+
+  /**
+   * Returns the {@link IndexSnapshot} whose {@link IndexSnapshot#sourceTableSnapshotId()} matches
+   * the given table snapshot ID, or null if none exists.
+   */
+  default IndexSnapshot snapshotForTableSnapshot(long tableSnapshotId) {
+    for (IndexSnapshot snap : snapshots()) {
+      if (snap.sourceTableSnapshotId() == tableSnapshotId) {
+        return snap;
+      }
+    }
+    return null;
+  }
+}

api/src/main/java/org/apache/iceberg/index/IndexSnapshot.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.index;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+/**
+ * An immutable version of the index data corresponding to a specific source table snapshot.
+ *
+ * <p>Each index snapshot is tied to one table snapshot via {@link #sourceTableSnapshotId()}. The
+ * tracking file referenced by {@link #trackingFile()} lists all leaf files belonging to this
+ * snapshot.
+ */
+public interface IndexSnapshot {
+
+  /** Unique identifier for this index snapshot. */
+  long snapshotId();
+
+  /** The source table snapshot this index snapshot was built from. */
+  long sourceTableSnapshotId();
+
+  /** Timestamp in milliseconds when this index snapshot was created. */
+  long timestampMs();
+
+  /**
+   * Location of the tracking file for this index snapshot.
+   *
+   * <p>The tracking file is a Parquet file that lists all leaf files belonging to this snapshot,
+   * with transform-value min/max bounds per leaf file for planning-time pruning.
+   */
+  String trackingFile();
+
+  /**
+   * Optional snapshot-level properties, supplied by the index maintenance process.
+   *
+   * @return an unmodifiable map of string properties, never null
+   */
+  Map<String, String> properties();
+
+  /**
+   * Optional implementation-specific key metadata for tracking file encryption.
+   *
+   * @return key metadata bytes, or null if not set
+   */
+  @Nullable
+  ByteBuffer keyMetadata();
+}