From c06d049a540df1a61395d2564dbd25732ecc0310 Mon Sep 17 00:00:00 2001 From: Manish Malhotra Date: Mon, 20 Jul 2026 13:57:51 -0700 Subject: [PATCH 1/5] Spec: Add secondary index specification (from PR #16961) Applies the index spec document from apache/iceberg PR #16961 authored by pvary. This defines the generic secondary index framework: SCALAR index type, HASH/IDENTITY transforms, Index Metadata, Tracking File, and Leaf File structure. --- format/index.md | 390 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 390 insertions(+) create mode 100644 format/index.md diff --git a/format/index.md b/format/index.md new file mode 100644 index 000000000000..c2e4448cd760 --- /dev/null +++ b/format/index.md @@ -0,0 +1,390 @@ +--- +title: "Index Spec" +--- + + +# Iceberg Index Specification + +## Background and Motivation + +Indexes enable query engines to locate relevant rows without scanning entire datasets. +They can accelerate point lookups, range predicates, and other retrieval patterns +while preserving Iceberg's table format, snapshot isolation, and interoperability. + +Indexes are optional. Engines may choose to create, maintain, consume, or ignore them. + +## Goals + +- Define a portable metadata format for indexes +- Provide a common storage architecture for index data +- Allow indexes to be operated independently from source table metadata +- Enable index sharing across engines +- Provide a framework for defining new index types and transform functions + +## Overview + +Indexes are stored as a collection of files with some Iceberg table like semantics. At a high level they consist of a tracking file (similar to a root manifest file) which contains listings for a defined set of leaf files (similar to data files). Leaf files store an ordered set of rows, each containing at least a key, the path of the Iceberg table data file, and the position within that file where the row for that key is stored. The organization of leaf files is defined by an Index Transform Function which varies based on the type of index. This structure is recorded in an Index metadata.json file which contains a set of snapshots, each of which points to a single tracking file mapping to the complete state of an Iceberg table at a given Iceberg table snapshot. + +Like Iceberg tables, views, and functions: + +- Metadata files (index metadata and tracking files) and data files (leaf files) are immutable +- Updates create new metadata files +- Catalogs perform atomic metadata swaps + +Each index snapshot references a tracking file which describes the leaf files belonging to the snapshot. + +```text +Index Metadata + | + +-- Index Snapshot(s) + | + +-- Tracking File + | + +-- Leaf Data Files +``` + +Transform functions derive a transform value from the key columns and determine how index entries are organized within +the leaf files. +- The transform value space is divided into non-overlapping ranges. +- Each leaf file stores entries for a single range. +- The tracking file stores range bounds for each leaf file. + +This structure enables efficient planning while keeping the data layout flexible for different index implementations. + +## Definitions + +### Index Type + +The index type defines the logical category of an index and the class of queries it is designed to accelerate. + +The metadata, snapshot, tracking-file, and leaf-file structures defined in this specification form a generic framework shared by all index types. Each index type builds on this framework by defining its type-specific details, such as the leaf schema and the applicable transform functions. + +The following index type is defined in this specification: + +| Type | Description | +|--------|------------------------------------------------------------------| +| SCALAR | Maps scalar key values to their locations for equality lookups. | + +The following index types are reserved for future specifications. Their identifiers are claimed so that engines and catalog implementations recognize them as valid type names and handle them gracefully, but this specification defines no type-specific requirements (leaf schema, transforms, or query semantics) for them: + +| Type | Description | +|--------|----------------------------------------------------------| +| VECTOR | Reserved for similarity search over vector embeddings. | +| TERM | Reserved for text/term search. | + +The index type communicates the capabilities of an index to query engines and helps determine whether an index is +applicable to a particular query. + +### Index Transform Function + +The index transform function defines how the transform value is derived from the key columns when rows are stored in the +index. The following terms are used throughout this specification: + +- **Key columns**: the source-table columns the transform function is applied to. +- **Transform value**: the value produced by applying the transform function to a row's key columns. Index entries are organized by transform value. +- **Included columns**: optional source-table columns copied into the index for read convenience. They do not affect how the index is organized. + +The transform function determines the physical organization of the indexed data and therefore influences which query +patterns can efficiently leverage the index. + +The following transform functions are defined in this specification. The bound interpretation describes what the +transform-value bounds stored in the tracking file represent for each transform: + +| Transform | Bound Interpretation | +|-----------|----------------------| +| IDENTITY | Original value range | +| HASH | Hash bucket range | + +The following transform functions are reserved for future specifications: + +| Transform | Bound Interpretation | +|-----------|---------------------------| +| HILBERT | Hilbert key range | +| IVF | Centroid identifier range | + +An index type does not fix a single transform function; the same index type can be realized with different transform functions. + +### Index Instance + +An index instance is a concrete realization of an index type and function applied to a specific table. + +Users create index instances by specifying: + +- The source table +- The index type +- The transform function +- The key columns +- The included columns (optional) +- Index properties (optional) + +Multiple instances of the same index type may exist for a table. + +### Index Snapshot + +An index snapshot is an immutable version of the index data generated from a specific table snapshot. + +Each index snapshot references a complete set of index files and contains all data from the referenced table snapshot. + +## Index Metadata + +The index metadata file stores the index definition and snapshot history. + +### Index Metadata File + +| Requirement | Field | Type | Description | +|-------------|---------------------|--------------------------|-------------------------------------------------| +| required | format-version | int | Index specification version | +| required | uuid | string | Stable UUID assigned at creation | +| required | table-uuid | string | UUID of the indexed table | +| required | location | string | Index root location | +| required | type | string | Logical index type | +| required | transform-function | string | Physical organization transform | +| required | key-column-ids | list | Source-table column IDs the transform is applied to (key columns) | +| optional | included-column-ids | list | Source-table column IDs copied into the index for read convenience (included columns) | +| optional | properties | map | Index properties applicable for every snapshot | +| required | snapshots | list | Index snapshots | + +## Index Snapshot + +Each index snapshot corresponds to one version of the index data. + +| Requirement | Field | Type | Description | +|-------------|--------------------------|--------------------|---------------------------------------------------------------------| +| required | snapshot-id | long | Index snapshot identifier | +| required | source-table-snapshot-id | long | Source table snapshot | +| required | timestamp-ms | long | Snapshot creation timestamp | +| required | tracking-file | string | Tracking file location | +| optional | properties | map | Snapshot properties specific to this snapshot | +| optional | key-metadata | binary | Implementation-specific key metadata, for tracking file encryption. | + +## Tracking File + +Each index snapshot references exactly one tracking file. + +It contains summary metadata about all leaf files belonging to the index snapshot and enables efficient planning +without scanning every leaf file. + +The tracking file may be stored using any supported metadata file format. + +### Tracking File Entry + +Each tracking file contains a collection of tracking file entries. A tracking file entry describes a single leaf file +tracked by an index snapshot. The fields are the subset of the V4 manifest entry fields that are relevant to planning +queries against the index. +Entries contain aggregated statistics for all referenced leaf files, enabling engines to perform pruning and planning +without opening every leaf file. + +| Field ID | Name | Type | Requirement | Description | +|----------|--------------------|---------|--------------|------------------------------------------------------------------------------------------------------------------------| +| 100 | location | string | required | Location of the referenced file. | +| 101 | file_format | string | required | File format name, such as parquet, avro, or orc. | +| 103 | record_count | long | required | Number of records contained in the referenced leaf file. | +| 104 | file_size_in_bytes | long | required | Total file size in bytes. | +| 146 | content_stats | struct | required | Statistics used for planning and pruning, including transform_value min/max statistics and optional column statistics. | +| 131 | key_metadata | binary | optional | Implementation-specific key metadata, used for leaf file encryption. | + +### Content Statistics + +The content statistics structure contains transform-key statistics and optional column statistics for the referenced +file. The transform-key statistics are always present, while column statistics are optional and may be omitted for +performance reasons. + +## Leaf Files + +Leaf files contain the actual index entries and represent the lowest level of the index hierarchy. + +Leaf files must be standard Iceberg data files and may be stored using any Iceberg-supported file format: +- Parquet +- Avro +- ORC - May be removed if ORC support is deprecated in Iceberg. + +The schema of a leaf file is determined by the index definition and contains: +- All key columns defined by the index +- Any included columns defined by the index +- The transform value produced by the transform function +- The source file path +- The source row position + +Entries within a leaf file are sorted by transform value, then by the key columns. This lets a reader binary-search for a +key within the leaf after selecting it from the tracking file, and groups all entries that share a key. + +### Transform value ranges + +This section describes how transform values organize leaf files. For the list of transform functions and their bound +interpretations, see [Index Transform Function](#index-transform-function). + +The transform function produces a transform value for each indexed row. To enable efficient planning, the transform +value space is divided into non-overlapping ranges. Each leaf file contains entries for a single range, while the +tracking file stores the corresponding bounds for every leaf file. If, and only if, a single transform value produces +more rows than fit in one leaf file, multiple leaf files may be created for that value. + +When a query predicate can be mapped to transform value ranges, engines can use these bounds to prune leaf files that +cannot contain matching entries, avoiding unnecessary reads. + +The effectiveness of this pruning depends on the transform and the query pattern. A transform that preserves locality +between the source columns and the resulting transform value enables additional pruning using column statistics stored +in the tracking file. For example, a Hilbert transform can cluster similar multi-column keys together, which can reduce +the number of leaf files read for range scans and partial-key lookups. + +### Leaf Schema + +Columns originating from the source table must preserve their original Iceberg field identifiers. +Reusing the original field IDs ensures that schema evolution, column renames, and type compatibility semantics remain +consistent between the table and the index. + +The index-specific columns are: + +| Field Id | Column | Type | Description | +|-----------|-----------------|--------|------------------------------------------------------------------------| +| TBD | transform_value | long | The result of applying the index transform function to the key columns | +| TBD | file_path | string | The path of the source data file the entry references | +| TBD | position | long | The row position of the entry within the source data file | + +## Commits and Concurrency + +Index metadata is immutable. Every update, whether adding a new snapshot, dropping an old one, or changing index +properties, produces a new index metadata file rather than modifying the existing one. Each new metadata file is written +with a unique name. + +A commit replaces the current index metadata file with the new one. This swap is atomic: it succeeds only if the current index +metadata file is still the one the writer started from. The check is based on the current index metadata file name. +If, between the time a writer reads the metadata and the time it attempts to commit, another process has already +committed a newer index metadata file, the expected current file no longer matches and the commit is rejected. + +When a commit is rejected because of such a conflict, the writer does not overwrite the newer metadata. Instead, the index +maintenance process decides how to proceed. Depending on the situation it may: + +- Re-read the latest committed metadata and retry the update on top of it, or +- Discard the attempted update, for example when the conflicting commit already achieved the intended result. + +This prevents concurrent index maintenance runs from silently overwriting each other and losing snapshots. + +## Example: Key Lookup Index + +Imagine an `events` table that already has a single snapshot (source table snapshot `3055729675574597004`). To speed up +point lookups on the `user_id` column, a key lookup index is created. + +```sql +CREATE INDEX hash_index + ON events (user_id) + USING HASH; +``` + +This creates a `SCALAR` index that applies the `HASH` transform to the `user_id` key column. When the index is created, +the engine (or a later index maintenance job) reads the current table snapshot, writes the leaf files and a tracking +file, and produces the first index metadata file containing a single index snapshot. Leaf files are organized by +transform value, while the tracking file stores summary information and pruning statistics. + +Each leaf file row contains the key column, the transform value, and the location of the source row: + +| Column | Description | +|-----------------|------------------------------------------------------| +| user_id | The indexed key column | +| transform_value | The result of applying `HASH` to `user_id` | +| file_path | The source data file that contains the row | +| position | The row position within the source data file | + +The JSON metadata file is shown below. + +``` +s3://bucket/warehouse/default.db/events/index/hash_index/metadata/00001-(uuid).metadata.json +``` +```json +{ + "format-version" : 1, + "uuid" : "9c12d441-03fe-4693-9a96-a0705ddf69c1", + "table-uuid" : "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94", + "location" : "s3://bucket/warehouse/default.db/events/index/hash_index", + "type" : "SCALAR", + "transform-function" : "HASH", + "key-column-ids" : [ 1 ], + "snapshots" : [ { + "snapshot-id" : 1, + "source-table-snapshot-id" : 3055729675574597004, + "timestamp-ms" : 1573518431292, + "tracking-file" : "s3://bucket/warehouse/default.db/events/index/hash_index/metadata/tracking-00001-(uuid).parquet" + } ] +} +``` + +Later, new data is added to the `events` table, producing a new table snapshot (`5459876531255530170`). Index +maintenance runs again and writes new leaf files for the added data, plus a new tracking file that references both the +still-valid old leaf files and the new leaf files. + +This produces a new index metadata file that completely replaces the previous one. The old index snapshot (`snapshot-id` +1) is kept alongside the new one (`snapshot-id` 2), so engines can still use the index against the older table snapshot. + +``` +s3://bucket/warehouse/default.db/events/index/hash_index/metadata/00002-(uuid).metadata.json +``` +```json +{ + "format-version" : 1, + "uuid" : "9c12d441-03fe-4693-9a96-a0705ddf69c1", + "table-uuid" : "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94", + "location" : "s3://bucket/warehouse/default.db/events/index/hash_index", + "type" : "SCALAR", + "transform-function" : "HASH", + "key-column-ids" : [ 1 ], + "snapshots" : [ { + "snapshot-id" : 1, + "source-table-snapshot-id" : 3055729675574597004, + "timestamp-ms" : 1573518431292, + "tracking-file" : "s3://bucket/warehouse/default.db/events/index/hash_index/metadata/tracking-00001-(uuid).parquet" + }, { + "snapshot-id" : 2, + "source-table-snapshot-id" : 5459876531255530170, + "timestamp-ms" : 1573518981593, + "tracking-file" : "s3://bucket/warehouse/default.db/events/index/hash_index/metadata/tracking-00002-(uuid).parquet" + } ] +} +``` + +Eventually the older table snapshot is no longer needed, so maintenance drops the corresponding index snapshot +(`snapshot-id` 1). It writes a new index metadata file that removes the snapshot from the `snapshots` list and replaces +the previous metadata file. Maintenance then deletes the files referenced only by the removed snapshot: its tracking file +and any leaf files not referenced by a remaining snapshot. Leaf files still referenced by the remaining snapshot are +retained. + +``` +s3://bucket/warehouse/default.db/events/index/hash_index/metadata/00003-(uuid).metadata.json +``` +```json +{ + "format-version" : 1, + "uuid" : "9c12d441-03fe-4693-9a96-a0705ddf69c1", + "table-uuid" : "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94", + "location" : "s3://bucket/warehouse/default.db/events/index/hash_index", + "type" : "SCALAR", + "transform-function" : "HASH", + "key-column-ids" : [ 1 ], + "snapshots" : [ { + "snapshot-id" : 2, + "source-table-snapshot-id" : 5459876531255530170, + "timestamp-ms" : 1573518981593, + "tracking-file" : "s3://bucket/warehouse/default.db/events/index/hash_index/metadata/tracking-00002-(uuid).parquet" + } ] +} +``` + +## Future Extensions + +Future specifications may define: + +- VECTOR indexes +- TERM indexes From 13de9d194301c025f71d0619d3da5bdbb054bee1 Mon Sep 17 00:00:00 2001 From: Manish Malhotra Date: Mon, 20 Jul 2026 14:32:45 -0700 Subject: [PATCH 2/5] Core: Add IndexMetadata and IndexSnapshot for secondary index spec Implements IndexMetadata and IndexSnapshot POJOs matching the field definitions in format/index.md (PR #16961). Includes JSON parsers and a unit test suite covering round-trip serialization, snapshot lookup helpers, builder validation, and spec field name verification. 6/6 unit tests passing. --- .../apache/iceberg/index/IndexMetadata.java | 140 +++++++++ .../apache/iceberg/index/IndexSnapshot.java | 65 ++++ .../iceberg/index/GenericIndexMetadata.java | 278 ++++++++++++++++++ .../iceberg/index/GenericIndexSnapshot.java | 134 +++++++++ .../iceberg/index/IndexMetadataParser.java | 159 ++++++++++ .../iceberg/index/IndexSnapshotParser.java | 75 +++++ .../index/TestIndexMetadataParser.java | 188 ++++++++++++ 7 files changed, 1039 insertions(+) create mode 100644 api/src/main/java/org/apache/iceberg/index/IndexMetadata.java create mode 100644 api/src/main/java/org/apache/iceberg/index/IndexSnapshot.java create mode 100644 core/src/main/java/org/apache/iceberg/index/GenericIndexMetadata.java create mode 100644 core/src/main/java/org/apache/iceberg/index/GenericIndexSnapshot.java create mode 100644 core/src/main/java/org/apache/iceberg/index/IndexMetadataParser.java create mode 100644 core/src/main/java/org/apache/iceberg/index/IndexSnapshotParser.java create mode 100644 core/src/test/java/org/apache/iceberg/index/TestIndexMetadataParser.java diff --git a/api/src/main/java/org/apache/iceberg/index/IndexMetadata.java b/api/src/main/java/org/apache/iceberg/index/IndexMetadata.java new file mode 100644 index 000000000000..6bc2d3efd7fb --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/index/IndexMetadata.java @@ -0,0 +1,140 @@ +/* + * 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. + * + *

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. + * + *

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"}. + * + *

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"}. + * + *

Determines how index entries are organized within leaf files. + */ + String transformFunction(); + + /** + * Source-table column IDs that the transform is applied to (key columns). + * + *

Leaf files are organized by the transform value derived from these columns. + */ + List keyColumnIds(); + + /** + * Optional source-table column IDs copied into the index for read convenience (included + * columns). + * + *

These columns are stored in leaf files but do not affect index organization. Used to serve + * covering queries without reading the source table. + */ + List 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 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 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; + } +} diff --git a/api/src/main/java/org/apache/iceberg/index/IndexSnapshot.java b/api/src/main/java/org/apache/iceberg/index/IndexSnapshot.java new file mode 100644 index 000000000000..874be46dbbde --- /dev/null +++ b/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. + * + *

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. + * + *

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 properties(); + + /** + * Optional implementation-specific key metadata for tracking file encryption. + * + * @return key metadata bytes, or null if not set + */ + @Nullable + ByteBuffer keyMetadata(); +} diff --git a/core/src/main/java/org/apache/iceberg/index/GenericIndexMetadata.java b/core/src/main/java/org/apache/iceberg/index/GenericIndexMetadata.java new file mode 100644 index 000000000000..484b4703ac53 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/GenericIndexMetadata.java @@ -0,0 +1,278 @@ +/* + * 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 java.util.UUID; +import javax.annotation.Nullable; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; + +/** Immutable implementation of {@link IndexMetadata}. */ +public class GenericIndexMetadata implements IndexMetadata { + + private final int formatVersion; + private final String uuid; + private final String tableUuid; + private final String location; + private final String type; + private final String transformFunction; + private final List keyColumnIds; + private final List includedColumnIds; + private final Map properties; + private final Long currentSnapshotId; + private final List snapshots; + private final String metadataFileLocation; + + private GenericIndexMetadata( + int formatVersion, + String uuid, + String tableUuid, + String location, + String type, + String transformFunction, + List keyColumnIds, + List includedColumnIds, + Map properties, + Long currentSnapshotId, + List snapshots, + String metadataFileLocation) { + this.formatVersion = formatVersion; + this.uuid = uuid; + this.tableUuid = tableUuid; + this.location = location; + this.type = type; + this.transformFunction = transformFunction; + this.keyColumnIds = ImmutableList.copyOf(keyColumnIds); + this.includedColumnIds = + includedColumnIds != null ? ImmutableList.copyOf(includedColumnIds) : ImmutableList.of(); + this.properties = properties != null ? ImmutableMap.copyOf(properties) : ImmutableMap.of(); + this.currentSnapshotId = currentSnapshotId; + this.snapshots = ImmutableList.copyOf(snapshots); + this.metadataFileLocation = metadataFileLocation; + } + + @Override + public int formatVersion() { + return formatVersion; + } + + @Override + public String uuid() { + return uuid; + } + + @Override + public String tableUuid() { + return tableUuid; + } + + @Override + public String location() { + return location; + } + + @Override + public String type() { + return type; + } + + @Override + public String transformFunction() { + return transformFunction; + } + + @Override + public List keyColumnIds() { + return keyColumnIds; + } + + @Override + public List includedColumnIds() { + return includedColumnIds; + } + + @Override + public Map properties() { + return properties; + } + + @Override + @Nullable + public Long currentSnapshotId() { + return currentSnapshotId; + } + + @Override + public List snapshots() { + return snapshots; + } + + @Override + @Nullable + public String metadataFileLocation() { + return metadataFileLocation; + } + + public static Builder builder() { + return new Builder(); + } + + public static Builder buildFrom(IndexMetadata base) { + return new Builder(base); + } + + /** Builder for {@link GenericIndexMetadata}. */ + public static class Builder { + private int formatVersion = DEFAULT_INDEX_FORMAT_VERSION; + private String uuid; + private String tableUuid; + private String location; + private String type; + private String transformFunction; + private List keyColumnIds = ImmutableList.of(); + private List includedColumnIds = ImmutableList.of(); + private Map properties = ImmutableMap.of(); + private Long currentSnapshotId; + private final List snapshots; + private String metadataFileLocation; + + private Builder() { + this.uuid = UUID.randomUUID().toString(); + this.snapshots = Lists.newArrayList(); + } + + private Builder(IndexMetadata base) { + this.formatVersion = base.formatVersion(); + this.uuid = base.uuid(); + this.tableUuid = base.tableUuid(); + this.location = base.location(); + this.type = base.type(); + this.transformFunction = base.transformFunction(); + this.keyColumnIds = ImmutableList.copyOf(base.keyColumnIds()); + this.includedColumnIds = ImmutableList.copyOf(base.includedColumnIds()); + this.properties = ImmutableMap.copyOf(base.properties()); + this.currentSnapshotId = base.currentSnapshotId(); + this.snapshots = Lists.newArrayList(base.snapshots()); + } + + public Builder formatVersion(int version) { + Preconditions.checkArgument( + version > 0 && version <= SUPPORTED_INDEX_FORMAT_VERSION, + "Unsupported format version: %s", + version); + this.formatVersion = version; + return this; + } + + public Builder uuid(String indexUuid) { + this.uuid = Preconditions.checkNotNull(indexUuid, "uuid is required"); + return this; + } + + public Builder tableUuid(String tblUuid) { + this.tableUuid = Preconditions.checkNotNull(tblUuid, "table-uuid is required"); + return this; + } + + public Builder location(String loc) { + this.location = Preconditions.checkNotNull(loc, "location is required"); + return this; + } + + public Builder type(String indexType) { + this.type = Preconditions.checkNotNull(indexType, "type is required"); + return this; + } + + public Builder transformFunction(String transform) { + this.transformFunction = + Preconditions.checkNotNull(transform, "transform-function is required"); + return this; + } + + public Builder keyColumnIds(List ids) { + Preconditions.checkArgument( + ids != null && !ids.isEmpty(), "key-column-ids must be non-empty"); + this.keyColumnIds = ImmutableList.copyOf(ids); + return this; + } + + public Builder includedColumnIds(List ids) { + this.includedColumnIds = ids != null ? ImmutableList.copyOf(ids) : ImmutableList.of(); + return this; + } + + public Builder properties(Map props) { + this.properties = props != null ? ImmutableMap.copyOf(props) : ImmutableMap.of(); + return this; + } + + public Builder addSnapshot(IndexSnapshot snapshot) { + Preconditions.checkNotNull(snapshot, "snapshot is required"); + snapshots.add(snapshot); + this.currentSnapshotId = snapshot.snapshotId(); + return this; + } + + public Builder removeSnapshot(long snapshotId) { + snapshots.removeIf(s -> s.snapshotId() == snapshotId); + if (Long.valueOf(snapshotId).equals(currentSnapshotId)) { + this.currentSnapshotId = + snapshots.isEmpty() ? null : snapshots.get(snapshots.size() - 1).snapshotId(); + } + return this; + } + + public Builder currentSnapshotId(Long id) { + this.currentSnapshotId = id; + return this; + } + + public Builder metadataFileLocation(String path) { + this.metadataFileLocation = path; + return this; + } + + public GenericIndexMetadata build() { + Preconditions.checkArgument(uuid != null, "uuid is required"); + Preconditions.checkArgument(tableUuid != null, "table-uuid is required"); + Preconditions.checkArgument(location != null, "location is required"); + Preconditions.checkArgument(type != null, "type is required"); + Preconditions.checkArgument(transformFunction != null, "transform-function is required"); + Preconditions.checkArgument(!keyColumnIds.isEmpty(), "key-column-ids must be non-empty"); + return new GenericIndexMetadata( + formatVersion, + uuid, + tableUuid, + location, + type, + transformFunction, + keyColumnIds, + includedColumnIds, + properties, + currentSnapshotId, + snapshots, + metadataFileLocation); + } + } +} diff --git a/core/src/main/java/org/apache/iceberg/index/GenericIndexSnapshot.java b/core/src/main/java/org/apache/iceberg/index/GenericIndexSnapshot.java new file mode 100644 index 000000000000..5b33e4c6d28e --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/GenericIndexSnapshot.java @@ -0,0 +1,134 @@ +/* + * 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; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; + +/** Immutable implementation of {@link IndexSnapshot}. */ +public class GenericIndexSnapshot implements IndexSnapshot { + + private final long snapshotId; + private final long sourceTableSnapshotId; + private final long timestampMs; + private final String trackingFile; + private final Map properties; + private final ByteBuffer keyMetadata; + + private GenericIndexSnapshot( + long snapshotId, + long sourceTableSnapshotId, + long timestampMs, + String trackingFile, + Map properties, + ByteBuffer keyMetadata) { + this.snapshotId = snapshotId; + this.sourceTableSnapshotId = sourceTableSnapshotId; + this.timestampMs = timestampMs; + this.trackingFile = trackingFile; + this.properties = properties != null ? ImmutableMap.copyOf(properties) : ImmutableMap.of(); + this.keyMetadata = keyMetadata; + } + + @Override + public long snapshotId() { + return snapshotId; + } + + @Override + public long sourceTableSnapshotId() { + return sourceTableSnapshotId; + } + + @Override + public long timestampMs() { + return timestampMs; + } + + @Override + public String trackingFile() { + return trackingFile; + } + + @Override + public Map properties() { + return properties; + } + + @Override + @Nullable + public ByteBuffer keyMetadata() { + return keyMetadata; + } + + public static Builder builder() { + return new Builder(); + } + + public static class Builder { + private long snapshotId; + private long sourceTableSnapshotId; + private long timestampMs; + private String trackingFile; + private Map properties; + private ByteBuffer keyMetadata; + + private Builder() {} + + public Builder snapshotId(long id) { + this.snapshotId = id; + return this; + } + + public Builder sourceTableSnapshotId(long id) { + this.sourceTableSnapshotId = id; + return this; + } + + public Builder timestampMs(long ts) { + this.timestampMs = ts; + return this; + } + + public Builder trackingFile(String path) { + this.trackingFile = path; + return this; + } + + public Builder properties(Map props) { + this.properties = props; + return this; + } + + public Builder keyMetadata(ByteBuffer metadata) { + this.keyMetadata = metadata; + return this; + } + + public GenericIndexSnapshot build() { + Preconditions.checkArgument(trackingFile != null, "tracking-file is required"); + Preconditions.checkArgument(timestampMs > 0, "timestamp-ms is required"); + return new GenericIndexSnapshot( + snapshotId, sourceTableSnapshotId, timestampMs, trackingFile, properties, keyMetadata); + } + } +} diff --git a/core/src/main/java/org/apache/iceberg/index/IndexMetadataParser.java b/core/src/main/java/org/apache/iceberg/index/IndexMetadataParser.java new file mode 100644 index 000000000000..c0460c3b0511 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/IndexMetadataParser.java @@ -0,0 +1,159 @@ +/* + * 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 com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.databind.JsonNode; +import java.io.IOException; +import java.io.StringWriter; +import java.io.UncheckedIOException; +import java.util.Iterator; +import java.util.List; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.util.JsonUtil; + +/** + * JSON serialization for {@link IndexMetadata}. + * + *

The format matches the index metadata JSON defined in format/index.md. + */ +public class IndexMetadataParser { + + private static final String FORMAT_VERSION = "format-version"; + private static final String UUID = "uuid"; + private static final String TABLE_UUID = "table-uuid"; + private static final String LOCATION = "location"; + private static final String TYPE = "type"; + private static final String TRANSFORM_FUNCTION = "transform-function"; + private static final String KEY_COLUMN_IDS = "key-column-ids"; + private static final String INCLUDED_COLUMN_IDS = "included-column-ids"; + private static final String PROPERTIES = "properties"; + private static final String CURRENT_SNAPSHOT_ID = "current-snapshot-id"; + private static final String SNAPSHOTS = "snapshots"; + + private IndexMetadataParser() {} + + public static String toJson(IndexMetadata metadata) { + return toJson(metadata, false); + } + + public static String toJson(IndexMetadata metadata, boolean pretty) { + try { + StringWriter writer = new StringWriter(); + JsonGenerator generator = JsonUtil.factory().createGenerator(writer); + if (pretty) { + generator.useDefaultPrettyPrinter(); + } + toJson(metadata, generator); + generator.flush(); + return writer.toString(); + } catch (IOException e) { + throw new UncheckedIOException("Failed to serialize index metadata", e); + } + } + + static void toJson(IndexMetadata metadata, JsonGenerator generator) throws IOException { + generator.writeStartObject(); + generator.writeNumberField(FORMAT_VERSION, metadata.formatVersion()); + generator.writeStringField(UUID, metadata.uuid()); + generator.writeStringField(TABLE_UUID, metadata.tableUuid()); + generator.writeStringField(LOCATION, metadata.location()); + generator.writeStringField(TYPE, metadata.type()); + generator.writeStringField(TRANSFORM_FUNCTION, metadata.transformFunction()); + + generator.writeArrayFieldStart(KEY_COLUMN_IDS); + for (int id : metadata.keyColumnIds()) { + generator.writeNumber(id); + } + generator.writeEndArray(); + + if (!metadata.includedColumnIds().isEmpty()) { + generator.writeArrayFieldStart(INCLUDED_COLUMN_IDS); + for (int id : metadata.includedColumnIds()) { + generator.writeNumber(id); + } + generator.writeEndArray(); + } + + if (!metadata.properties().isEmpty()) { + JsonUtil.writeStringMap(PROPERTIES, metadata.properties(), generator); + } + + if (metadata.currentSnapshotId() != null) { + generator.writeNumberField(CURRENT_SNAPSHOT_ID, metadata.currentSnapshotId()); + } + + generator.writeArrayFieldStart(SNAPSHOTS); + for (IndexSnapshot snapshot : metadata.snapshots()) { + IndexSnapshotParser.toJson(snapshot, generator); + } + generator.writeEndArray(); + + generator.writeEndObject(); + } + + public static IndexMetadata fromJson(String json) { + try { + return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class)); + } catch (IOException e) { + throw new UncheckedIOException("Failed to deserialize index metadata", e); + } + } + + static IndexMetadata fromJson(JsonNode node) { + GenericIndexMetadata.Builder builder = GenericIndexMetadata.builder(); + builder.formatVersion(JsonUtil.getInt(FORMAT_VERSION, node)); + builder.uuid(JsonUtil.getString(UUID, node)); + builder.tableUuid(JsonUtil.getString(TABLE_UUID, node)); + builder.location(JsonUtil.getString(LOCATION, node)); + builder.type(JsonUtil.getString(TYPE, node)); + builder.transformFunction(JsonUtil.getString(TRANSFORM_FUNCTION, node)); + builder.keyColumnIds(JsonUtil.getIntegerList(KEY_COLUMN_IDS, node)); + + if (node.has(INCLUDED_COLUMN_IDS)) { + builder.includedColumnIds(JsonUtil.getIntegerList(INCLUDED_COLUMN_IDS, node)); + } + if (node.has(PROPERTIES)) { + builder.properties(JsonUtil.getStringMap(PROPERTIES, node)); + } + if (node.has(CURRENT_SNAPSHOT_ID) && !node.get(CURRENT_SNAPSHOT_ID).isNull()) { + builder.currentSnapshotId(JsonUtil.getLong(CURRENT_SNAPSHOT_ID, node)); + } + + if (node.has(SNAPSHOTS)) { + List snapshots = Lists.newArrayList(); + Iterator snapshotNodes = node.get(SNAPSHOTS).elements(); + while (snapshotNodes.hasNext()) { + snapshots.add(IndexSnapshotParser.fromJson(snapshotNodes.next())); + } + // add snapshots individually so builder tracks currentSnapshotId + for (IndexSnapshot snapshot : snapshots) { + builder.addSnapshot(snapshot); + } + // override currentSnapshotId with what was in JSON (addSnapshot sets it to last added) + if (node.has(CURRENT_SNAPSHOT_ID) && !node.get(CURRENT_SNAPSHOT_ID).isNull()) { + builder.currentSnapshotId(JsonUtil.getLong(CURRENT_SNAPSHOT_ID, node)); + } else if (snapshots.isEmpty()) { + builder.currentSnapshotId(null); + } + } + + return builder.build(); + } +} diff --git a/core/src/main/java/org/apache/iceberg/index/IndexSnapshotParser.java b/core/src/main/java/org/apache/iceberg/index/IndexSnapshotParser.java new file mode 100644 index 000000000000..1b407e221483 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/IndexSnapshotParser.java @@ -0,0 +1,75 @@ +/* + * 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 com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.databind.JsonNode; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Map; +import org.apache.iceberg.util.ByteBuffers; +import org.apache.iceberg.util.JsonUtil; + +/** JSON serialization for {@link IndexSnapshot}. */ +class IndexSnapshotParser { + + private static final String SNAPSHOT_ID = "snapshot-id"; + private static final String SOURCE_TABLE_SNAPSHOT_ID = "source-table-snapshot-id"; + private static final String TIMESTAMP_MS = "timestamp-ms"; + private static final String TRACKING_FILE = "tracking-file"; + private static final String PROPERTIES = "properties"; + private static final String KEY_METADATA = "key-metadata"; + + private IndexSnapshotParser() {} + + static void toJson(IndexSnapshot snapshot, JsonGenerator generator) throws IOException { + generator.writeStartObject(); + generator.writeNumberField(SNAPSHOT_ID, snapshot.snapshotId()); + generator.writeNumberField(SOURCE_TABLE_SNAPSHOT_ID, snapshot.sourceTableSnapshotId()); + generator.writeNumberField(TIMESTAMP_MS, snapshot.timestampMs()); + generator.writeStringField(TRACKING_FILE, snapshot.trackingFile()); + if (!snapshot.properties().isEmpty()) { + JsonUtil.writeStringMap(PROPERTIES, snapshot.properties(), generator); + } + if (snapshot.keyMetadata() != null) { + generator.writeBinaryField(KEY_METADATA, ByteBuffers.toByteArray(snapshot.keyMetadata())); + } + generator.writeEndObject(); + } + + static IndexSnapshot fromJson(JsonNode node) { + GenericIndexSnapshot.Builder builder = GenericIndexSnapshot.builder(); + builder.snapshotId(JsonUtil.getLong(SNAPSHOT_ID, node)); + builder.sourceTableSnapshotId(JsonUtil.getLong(SOURCE_TABLE_SNAPSHOT_ID, node)); + builder.timestampMs(JsonUtil.getLong(TIMESTAMP_MS, node)); + builder.trackingFile(JsonUtil.getString(TRACKING_FILE, node)); + if (node.has(PROPERTIES)) { + Map props = JsonUtil.getStringMap(PROPERTIES, node); + builder.properties(props); + } + if (node.has(KEY_METADATA)) { + try { + builder.keyMetadata(ByteBuffer.wrap(node.get(KEY_METADATA).binaryValue())); + } catch (IOException e) { + throw new RuntimeException("Failed to read key-metadata", e); + } + } + return builder.build(); + } +} diff --git a/core/src/test/java/org/apache/iceberg/index/TestIndexMetadataParser.java b/core/src/test/java/org/apache/iceberg/index/TestIndexMetadataParser.java new file mode 100644 index 000000000000..6de44d8f0348 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/index/TestIndexMetadataParser.java @@ -0,0 +1,188 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.junit.jupiter.api.Test; + +public class TestIndexMetadataParser { + + private static final String TABLE_UUID = "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94"; + private static final String INDEX_UUID = "9c12d441-03fe-4693-9a96-a0705ddf69c1"; + private static final String LOCATION = "s3://warehouse/db/orders/index/order_id_idx"; + private static final String TRACKING_FILE = + "s3://warehouse/db/orders/index/order_id_idx/metadata/tracking-00001.parquet"; + + @Test + public void roundTripNoSnapshots() { + IndexMetadata metadata = + GenericIndexMetadata.builder() + .uuid(INDEX_UUID) + .tableUuid(TABLE_UUID) + .location(LOCATION) + .type("SCALAR") + .transformFunction("HASH") + .keyColumnIds(ImmutableList.of(3)) + .properties(ImmutableMap.of("hash.num-buckets", "256")) + .build(); + + String json = IndexMetadataParser.toJson(metadata, true); + IndexMetadata restored = IndexMetadataParser.fromJson(json); + + assertThat(restored.formatVersion()).isEqualTo(1); + assertThat(restored.uuid()).isEqualTo(INDEX_UUID); + assertThat(restored.tableUuid()).isEqualTo(TABLE_UUID); + assertThat(restored.location()).isEqualTo(LOCATION); + assertThat(restored.type()).isEqualTo("SCALAR"); + assertThat(restored.transformFunction()).isEqualTo("HASH"); + assertThat(restored.keyColumnIds()).containsExactly(3); + assertThat(restored.includedColumnIds()).isEmpty(); + assertThat(restored.properties()).containsEntry("hash.num-buckets", "256"); + assertThat(restored.currentSnapshotId()).isNull(); + assertThat(restored.snapshots()).isEmpty(); + } + + @Test + public void roundTripWithSnapshot() { + IndexSnapshot snapshot = + GenericIndexSnapshot.builder() + .snapshotId(1L) + .sourceTableSnapshotId(3055729675574597004L) + .timestampMs(1735689600000L) + .trackingFile(TRACKING_FILE) + .properties(ImmutableMap.of("build.engine", "spark")) + .build(); + + IndexMetadata metadata = + GenericIndexMetadata.builder() + .uuid(INDEX_UUID) + .tableUuid(TABLE_UUID) + .location(LOCATION) + .type("SCALAR") + .transformFunction("HASH") + .keyColumnIds(ImmutableList.of(3)) + .addSnapshot(snapshot) + .build(); + + String json = IndexMetadataParser.toJson(metadata, true); + IndexMetadata restored = IndexMetadataParser.fromJson(json); + + assertThat(restored.snapshots()).hasSize(1); + assertThat(restored.currentSnapshotId()).isEqualTo(1L); + + IndexSnapshot restoredSnap = restored.snapshots().get(0); + assertThat(restoredSnap.snapshotId()).isEqualTo(1L); + assertThat(restoredSnap.sourceTableSnapshotId()).isEqualTo(3055729675574597004L); + assertThat(restoredSnap.timestampMs()).isEqualTo(1735689600000L); + assertThat(restoredSnap.trackingFile()).isEqualTo(TRACKING_FILE); + assertThat(restoredSnap.properties()).containsEntry("build.engine", "spark"); + } + + @Test + public void roundTripWithIncludedColumns() { + IndexMetadata metadata = + GenericIndexMetadata.builder() + .uuid(INDEX_UUID) + .tableUuid(TABLE_UUID) + .location(LOCATION) + .type("SCALAR") + .transformFunction("IDENTITY") + .keyColumnIds(ImmutableList.of(5)) + .includedColumnIds(ImmutableList.of(6, 7)) + .build(); + + String json = IndexMetadataParser.toJson(metadata); + IndexMetadata restored = IndexMetadataParser.fromJson(json); + + assertThat(restored.keyColumnIds()).containsExactly(5); + assertThat(restored.includedColumnIds()).containsExactly(6, 7); + assertThat(restored.transformFunction()).isEqualTo("IDENTITY"); + } + + @Test + public void snapshotLookupByTableSnapshotId() { + long tableSnapshotId = 3055729675574597004L; + IndexSnapshot snapshot = + GenericIndexSnapshot.builder() + .snapshotId(1L) + .sourceTableSnapshotId(tableSnapshotId) + .timestampMs(1735689600000L) + .trackingFile(TRACKING_FILE) + .build(); + + IndexMetadata metadata = + GenericIndexMetadata.builder() + .uuid(INDEX_UUID) + .tableUuid(TABLE_UUID) + .location(LOCATION) + .type("SCALAR") + .transformFunction("HASH") + .keyColumnIds(ImmutableList.of(3)) + .addSnapshot(snapshot) + .build(); + + assertThat(metadata.snapshotForTableSnapshot(tableSnapshotId)).isNotNull(); + assertThat(metadata.snapshotForTableSnapshot(tableSnapshotId).snapshotId()).isEqualTo(1L); + assertThat(metadata.snapshotForTableSnapshot(99999L)).isNull(); + } + + @Test + public void builderRequiresKeyColumnIds() { + assertThatThrownBy( + () -> + GenericIndexMetadata.builder() + .uuid(INDEX_UUID) + .tableUuid(TABLE_UUID) + .location(LOCATION) + .type("SCALAR") + .transformFunction("HASH") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("key-column-ids"); + } + + @Test + public void matchesSpecJsonFormat() { + // Verify the JSON output matches the field names defined in format/index.md + IndexMetadata metadata = + GenericIndexMetadata.builder() + .uuid(INDEX_UUID) + .tableUuid(TABLE_UUID) + .location(LOCATION) + .type("SCALAR") + .transformFunction("HASH") + .keyColumnIds(ImmutableList.of(1)) + .properties(ImmutableMap.of("hash.num-buckets", "256")) + .build(); + + String json = IndexMetadataParser.toJson(metadata); + assertThat(json).contains("\"format-version\""); + assertThat(json).contains("\"uuid\""); + assertThat(json).contains("\"table-uuid\""); + assertThat(json).contains("\"location\""); + assertThat(json).contains("\"type\""); + assertThat(json).contains("\"transform-function\""); + assertThat(json).contains("\"key-column-ids\""); + assertThat(json).contains("\"snapshots\""); + } +} From 581e7a1db5318445e4176c07c9129d099af99317 Mon Sep 17 00:00:00 2001 From: Manish Malhotra Date: Mon, 20 Jul 2026 14:48:38 -0700 Subject: [PATCH 3/5] Core: Add IndexCatalog, IndexMetadataIO, and InMemoryIndexCatalog Adds Layer 1 of the SCALAR index implementation: - IndexIdentifier: identifies an index by table + name - IndexCatalog: interface with create/load/update/drop/list operations with optimistic concurrency on updateIndex - IndexMetadataIO: read/write index metadata JSON via FileIO, with metadata file location naming (00001-{uuid}.metadata.json) - InMemoryIndexCatalog: thread-safe in-memory implementation for tests, includes CAS-based conflict detection on updateIndex 14/14 unit tests passing. --- .../apache/iceberg/index/IndexCatalog.java | 91 +++++++++ .../apache/iceberg/index/IndexIdentifier.java | 75 ++++++++ .../iceberg/index/InMemoryIndexCatalog.java | 128 +++++++++++++ .../apache/iceberg/index/IndexMetadataIO.java | 107 +++++++++++ .../index/TestInMemoryIndexCatalog.java | 173 ++++++++++++++++++ 5 files changed, 574 insertions(+) create mode 100644 api/src/main/java/org/apache/iceberg/index/IndexCatalog.java create mode 100644 api/src/main/java/org/apache/iceberg/index/IndexIdentifier.java create mode 100644 core/src/main/java/org/apache/iceberg/index/InMemoryIndexCatalog.java create mode 100644 core/src/main/java/org/apache/iceberg/index/IndexMetadataIO.java create mode 100644 core/src/test/java/org/apache/iceberg/index/TestInMemoryIndexCatalog.java diff --git a/api/src/main/java/org/apache/iceberg/index/IndexCatalog.java b/api/src/main/java/org/apache/iceberg/index/IndexCatalog.java new file mode 100644 index 000000000000..6051b1fc649a --- /dev/null +++ b/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. + * + *

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. + * + *

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. + * + *

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 listIndexes(TableIdentifier tableIdentifier); +} diff --git a/api/src/main/java/org/apache/iceberg/index/IndexIdentifier.java b/api/src/main/java/org/apache/iceberg/index/IndexIdentifier.java new file mode 100644 index 000000000000..c869b0c789b7 --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/index/IndexIdentifier.java @@ -0,0 +1,75 @@ +/* + * 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. + * + *

Format: {@code ..} + * + *

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; + } +} diff --git a/core/src/main/java/org/apache/iceberg/index/InMemoryIndexCatalog.java b/core/src/main/java/org/apache/iceberg/index/InMemoryIndexCatalog.java new file mode 100644 index 000000000000..bc8e48df9bb0 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/InMemoryIndexCatalog.java @@ -0,0 +1,128 @@ +/* + * 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.ConcurrentModificationException; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * In-memory implementation of {@link IndexCatalog}. + * + *

Stores index metadata in a thread-safe map. Suitable for unit tests and local development. + * Does not persist state across JVM restarts. + * + *

Optimistic concurrency: {@link #updateIndex} checks that the base metadata file location + * matches what is currently stored before replacing it. + */ +public class InMemoryIndexCatalog implements IndexCatalog { + + // Maps IndexIdentifier → current IndexMetadata + private final Map store = new ConcurrentHashMap<>(); + + @Override + public void createIndex(IndexIdentifier identifier, IndexMetadata metadata) { + Preconditions.checkNotNull(identifier, "identifier is required"); + Preconditions.checkNotNull(metadata, "metadata is required"); + Preconditions.checkArgument( + metadata.metadataFileLocation() != null, + "metadata must have a metadataFileLocation set before registering"); + + IndexMetadata existing = store.putIfAbsent(identifier, metadata); + if (existing != null) { + throw new AlreadyExistsException("Index already exists: %s", identifier); + } + } + + @Override + public IndexMetadata loadIndex(IndexIdentifier identifier) { + Preconditions.checkNotNull(identifier, "identifier is required"); + IndexMetadata metadata = store.get(identifier); + if (metadata == null) { + throw new NoSuchTableException("Index does not exist: %s", identifier); + } + return metadata; + } + + @Override + public void updateIndex( + IndexIdentifier identifier, IndexMetadata base, IndexMetadata updated) { + Preconditions.checkNotNull(identifier, "identifier is required"); + Preconditions.checkNotNull(base, "base metadata is required"); + Preconditions.checkNotNull(updated, "updated metadata is required"); + Preconditions.checkArgument( + updated.metadataFileLocation() != null, + "updated metadata must have a metadataFileLocation set"); + + // Optimistic concurrency: only update if the current location matches base + boolean replaced = + store.compute( + identifier, + (key, current) -> { + if (current == null) { + throw new NoSuchTableException("Index does not exist: %s", identifier); + } + if (!Objects.equals( + current.metadataFileLocation(), base.metadataFileLocation())) { + return null; // signal conflict + } + return updated; + }) + != null; + + if (!replaced) { + throw new ConcurrentModificationException( + String.format( + "Cannot update index %s: current metadata location has changed. " + + "Expected: %s", + identifier, base.metadataFileLocation())); + } + } + + @Override + public void dropIndex(IndexIdentifier identifier) { + Preconditions.checkNotNull(identifier, "identifier is required"); + IndexMetadata removed = store.remove(identifier); + if (removed == null) { + throw new NoSuchTableException("Index does not exist: %s", identifier); + } + } + + @Override + public boolean indexExists(IndexIdentifier identifier) { + Preconditions.checkNotNull(identifier, "identifier is required"); + return store.containsKey(identifier); + } + + @Override + public List listIndexes(TableIdentifier tableIdentifier) { + Preconditions.checkNotNull(tableIdentifier, "tableIdentifier is required"); + return store.entrySet().stream() + .filter(e -> e.getKey().tableIdentifier().equals(tableIdentifier)) + .map(Map.Entry::getValue) + .collect(Collectors.toList()); + } +} diff --git a/core/src/main/java/org/apache/iceberg/index/IndexMetadataIO.java b/core/src/main/java/org/apache/iceberg/index/IndexMetadataIO.java new file mode 100644 index 000000000000..7b41ebde3aa9 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/IndexMetadataIO.java @@ -0,0 +1,107 @@ +/* + * 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.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.util.UUID; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Reads and writes {@link IndexMetadata} JSON files using Iceberg's {@link FileIO}. + * + *

Metadata file naming follows the pattern: + * {@code {index_location}/metadata/{version:05d}-{uuid}.metadata.json} + * + *

Example: {@code s3://warehouse/db/orders/index/order_id_idx/metadata/00001-abc123.metadata.json} + */ +public class IndexMetadataIO { + + private IndexMetadataIO() {} + + /** + * Read index metadata from a file path. + * + * @param io the FileIO to use for reading + * @param metadataLocation the full path to the metadata JSON file + * @return the parsed IndexMetadata with metadataFileLocation set + */ + public static IndexMetadata read(FileIO io, String metadataLocation) { + Preconditions.checkNotNull(io, "FileIO is required"); + Preconditions.checkNotNull(metadataLocation, "metadataLocation is required"); + + InputFile inputFile = io.newInputFile(metadataLocation); + try (InputStream stream = inputFile.newStream()) { + String json = new String(stream.readAllBytes(), StandardCharsets.UTF_8); + IndexMetadata metadata = IndexMetadataParser.fromJson(json); + // Return with metadataFileLocation set so callers can use it for CAS + return GenericIndexMetadata.buildFrom(metadata) + .metadataFileLocation(metadataLocation) + .build(); + } catch (IOException e) { + throw new UncheckedIOException( + "Failed to read index metadata from: " + metadataLocation, e); + } + } + + /** + * Write index metadata to an OutputFile. + * + * @param metadata the index metadata to write + * @param outputFile the output file to write to + */ + public static void write(IndexMetadata metadata, OutputFile outputFile) { + Preconditions.checkNotNull(metadata, "metadata is required"); + Preconditions.checkNotNull(outputFile, "outputFile is required"); + + String json = IndexMetadataParser.toJson(metadata, true); + try (OutputStream stream = outputFile.createOrOverwrite()) { + stream.write(json.getBytes(StandardCharsets.UTF_8)); + } catch (IOException e) { + throw new UncheckedIOException( + "Failed to write index metadata to: " + outputFile.location(), e); + } + } + + /** + * Generate the next metadata file location for an index. + * + *

The version is derived from the number of existing snapshots + 1, ensuring monotonically + * increasing names for easy visual ordering. A UUID suffix guarantees uniqueness even if two + * writers generate the same version number concurrently. + * + * @param indexLocation the base location of the index + * @param currentMetadata the current metadata, or null if this is the first write + * @return the new metadata file path + */ + public static String newMetadataFileLocation( + String indexLocation, IndexMetadata currentMetadata) { + int version = currentMetadata == null ? 1 : currentMetadata.snapshots().size() + 1; + String uuid = UUID.randomUUID().toString().replace("-", "").substring(0, 8); + return String.format( + "%s/metadata/%05d-%s.metadata.json", + indexLocation.replaceAll("/$", ""), version, uuid); + } +} diff --git a/core/src/test/java/org/apache/iceberg/index/TestInMemoryIndexCatalog.java b/core/src/test/java/org/apache/iceberg/index/TestInMemoryIndexCatalog.java new file mode 100644 index 000000000000..3d337b7b8ed7 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/index/TestInMemoryIndexCatalog.java @@ -0,0 +1,173 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.util.ConcurrentModificationException; +import java.util.List; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class TestInMemoryIndexCatalog { + + private static final TableIdentifier TABLE = TableIdentifier.of(Namespace.of("db"), "orders"); + private static final IndexIdentifier IDX = + IndexIdentifier.of(TABLE, "order_id_idx"); + + private InMemoryIndexCatalog catalog; + + @BeforeEach + void setup() { + catalog = new InMemoryIndexCatalog(); + } + + private IndexMetadata sampleMetadata(String metadataLocation) { + return GenericIndexMetadata.builder() + .uuid("9c12d441-03fe-4693-9a96-a0705ddf69c1") + .tableUuid("fb072c92-a02b-11e9-ae9c-1bb7bc9eca94") + .location("s3://warehouse/db/orders/index/order_id_idx") + .type("SCALAR") + .transformFunction("HASH") + .keyColumnIds(ImmutableList.of(3)) + .metadataFileLocation(metadataLocation) + .build(); + } + + @Test + void createAndLoad() { + IndexMetadata metadata = sampleMetadata("s3://warehouse/.../metadata/00001-abc.metadata.json"); + catalog.createIndex(IDX, metadata); + + IndexMetadata loaded = catalog.loadIndex(IDX); + assertThat(loaded.uuid()).isEqualTo(metadata.uuid()); + assertThat(loaded.type()).isEqualTo("SCALAR"); + assertThat(loaded.metadataFileLocation()).isEqualTo(metadata.metadataFileLocation()); + } + + @Test + void createDuplicateThrows() { + catalog.createIndex(IDX, sampleMetadata("s3://.../00001.metadata.json")); + assertThatThrownBy( + () -> catalog.createIndex(IDX, sampleMetadata("s3://.../00001.metadata.json"))) + .isInstanceOf(AlreadyExistsException.class) + .hasMessageContaining("already exists"); + } + + @Test + void loadNonExistentThrows() { + assertThatThrownBy(() -> catalog.loadIndex(IDX)) + .isInstanceOf(NoSuchTableException.class) + .hasMessageContaining("does not exist"); + } + + @Test + void updateSucceeds() { + IndexMetadata v1 = sampleMetadata("s3://.../00001.metadata.json"); + catalog.createIndex(IDX, v1); + + IndexMetadata v2 = + GenericIndexMetadata.buildFrom(v1) + .metadataFileLocation("s3://.../00002.metadata.json") + .build(); + catalog.updateIndex(IDX, v1, v2); + + assertThat(catalog.loadIndex(IDX).metadataFileLocation()) + .isEqualTo("s3://.../00002.metadata.json"); + } + + @Test + void updateConflictThrows() { + IndexMetadata v1 = sampleMetadata("s3://.../00001.metadata.json"); + catalog.createIndex(IDX, v1); + + // Simulate another writer already committed v2 + IndexMetadata v2 = + GenericIndexMetadata.buildFrom(v1) + .metadataFileLocation("s3://.../00002.metadata.json") + .build(); + catalog.updateIndex(IDX, v1, v2); + + // Now our writer tries to commit based on stale v1 + IndexMetadata v2conflict = + GenericIndexMetadata.buildFrom(v1) + .metadataFileLocation("s3://.../00002-conflict.metadata.json") + .build(); + assertThatThrownBy(() -> catalog.updateIndex(IDX, v1, v2conflict)) + .isInstanceOf(ConcurrentModificationException.class) + .hasMessageContaining("metadata location has changed"); + } + + @Test + void dropAndExists() { + catalog.createIndex(IDX, sampleMetadata("s3://.../00001.metadata.json")); + assertThat(catalog.indexExists(IDX)).isTrue(); + + catalog.dropIndex(IDX); + assertThat(catalog.indexExists(IDX)).isFalse(); + assertThatThrownBy(() -> catalog.loadIndex(IDX)) + .isInstanceOf(NoSuchTableException.class); + } + + @Test + void listIndexes() { + TableIdentifier table2 = TableIdentifier.of(Namespace.of("db"), "items"); + IndexIdentifier idx2 = IndexIdentifier.of(TABLE, "created_at_idx"); + IndexIdentifier idx3 = IndexIdentifier.of(table2, "item_id_idx"); + + catalog.createIndex(IDX, sampleMetadata("s3://.../orders/00001.metadata.json")); + catalog.createIndex(idx2, sampleMetadata("s3://.../created_at/00001.metadata.json")); + catalog.createIndex(idx3, sampleMetadata("s3://.../items/00001.metadata.json")); + + List ordersIndexes = catalog.listIndexes(TABLE); + assertThat(ordersIndexes).hasSize(2); + + List itemsIndexes = catalog.listIndexes(table2); + assertThat(itemsIndexes).hasSize(1); + } + + @Test + void newMetadataFileLocationFormat() { + String location = "s3://warehouse/db/orders/index/order_id_idx"; + String path = IndexMetadataIO.newMetadataFileLocation(location, null); + assertThat(path).startsWith(location + "/metadata/00001-"); + assertThat(path).endsWith(".metadata.json"); + + // Second write with existing metadata with 1 snapshot gets version 2 + IndexMetadata withOneSnapshot = + GenericIndexMetadata.buildFrom(sampleMetadata("s3://x")) + .addSnapshot( + GenericIndexSnapshot.builder() + .snapshotId(1L) + .sourceTableSnapshotId(100L) + .timestampMs(System.currentTimeMillis()) + .trackingFile("s3://x/tracking.parquet") + .build()) + .build(); + String path2 = IndexMetadataIO.newMetadataFileLocation(location, withOneSnapshot); + assertThat(path2).startsWith(location + "/metadata/00002-"); + } +} From 44dc20277bb86ef5ac2ac90218aca96786390796 Mon Sep 17 00:00:00 2001 From: Manish Malhotra Date: Mon, 20 Jul 2026 17:33:30 -0700 Subject: [PATCH 4/5] Core: Add TrackingFileWriter and TrackingFileReader (Avro) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the tracking file format for the secondary index spec: - TrackingFileEntry: value class holding leaf file location, bounds, record count, size, and optional key metadata (field IDs from spec) - TrackingFileWriter: writes entries as an Avro DataFile using Snappy compression, writing to Iceberg OutputFile (same pattern as manifests) - TrackingFileReader: readAll() and readMatching(min, max) for planning-time pruning — returns only leaf files whose transform-value range overlaps the query range 18/18 unit tests passing. --- .../iceberg/index/TrackingFileEntry.java | 153 +++++++++++ .../iceberg/index/TrackingFileReader.java | 115 ++++++++ .../iceberg/index/TrackingFileWriter.java | 120 +++++++++ .../iceberg/index/TestTrackingFile.java | 247 ++++++++++++++++++ 4 files changed, 635 insertions(+) create mode 100644 core/src/main/java/org/apache/iceberg/index/TrackingFileEntry.java create mode 100644 core/src/main/java/org/apache/iceberg/index/TrackingFileReader.java create mode 100644 core/src/main/java/org/apache/iceberg/index/TrackingFileWriter.java create mode 100644 core/src/test/java/org/apache/iceberg/index/TestTrackingFile.java diff --git a/core/src/main/java/org/apache/iceberg/index/TrackingFileEntry.java b/core/src/main/java/org/apache/iceberg/index/TrackingFileEntry.java new file mode 100644 index 000000000000..774bcc4abf3b --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/TrackingFileEntry.java @@ -0,0 +1,153 @@ +/* + * 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 javax.annotation.Nullable; + +/** + * A single entry in a tracking file, describing one leaf file in an index snapshot. + * + *

The {@link #transformValueLowerBound()} and {@link #transformValueUpperBound()} fields are + * the key pruning statistics — the planner uses them to skip leaf files that cannot contain + * matching entries for a given predicate. + * + *

Field IDs match the tracking file entry schema defined in format/index.md. + */ +public class TrackingFileEntry { + + private final String location; // field 100 + private final String fileFormat; // field 101 + private final long recordCount; // field 103 + private final long fileSizeInBytes; // field 104 + private final long transformValueLowerBound; // field 146.lower + private final long transformValueUpperBound; // field 146.upper + private final ByteBuffer keyMetadata; // field 131 (optional) + + private TrackingFileEntry( + String location, + String fileFormat, + long recordCount, + long fileSizeInBytes, + long transformValueLowerBound, + long transformValueUpperBound, + ByteBuffer keyMetadata) { + this.location = location; + this.fileFormat = fileFormat; + this.recordCount = recordCount; + this.fileSizeInBytes = fileSizeInBytes; + this.transformValueLowerBound = transformValueLowerBound; + this.transformValueUpperBound = transformValueUpperBound; + this.keyMetadata = keyMetadata; + } + + public String location() { + return location; + } + + public String fileFormat() { + return fileFormat; + } + + public long recordCount() { + return recordCount; + } + + public long fileSizeInBytes() { + return fileSizeInBytes; + } + + /** Minimum transform value stored in this leaf file. Used for planning-time pruning. */ + public long transformValueLowerBound() { + return transformValueLowerBound; + } + + /** Maximum transform value stored in this leaf file. Used for planning-time pruning. */ + public long transformValueUpperBound() { + return transformValueUpperBound; + } + + @Nullable + public ByteBuffer keyMetadata() { + return keyMetadata; + } + + public static Builder builder() { + return new Builder(); + } + + public static class Builder { + private String location; + private String fileFormat = "parquet"; + private long recordCount; + private long fileSizeInBytes; + private long transformValueLowerBound; + private long transformValueUpperBound; + private ByteBuffer keyMetadata; + + public Builder location(String loc) { + this.location = loc; + return this; + } + + public Builder fileFormat(String format) { + this.fileFormat = format; + return this; + } + + public Builder recordCount(long count) { + this.recordCount = count; + return this; + } + + public Builder fileSizeInBytes(long size) { + this.fileSizeInBytes = size; + return this; + } + + public Builder transformValueLowerBound(long lower) { + this.transformValueLowerBound = lower; + return this; + } + + public Builder transformValueUpperBound(long upper) { + this.transformValueUpperBound = upper; + return this; + } + + public Builder keyMetadata(ByteBuffer metadata) { + this.keyMetadata = metadata; + return this; + } + + public TrackingFileEntry build() { + if (location == null || location.isEmpty()) { + throw new IllegalArgumentException("location is required"); + } + return new TrackingFileEntry( + location, + fileFormat, + recordCount, + fileSizeInBytes, + transformValueLowerBound, + transformValueUpperBound, + keyMetadata); + } + } +} diff --git a/core/src/main/java/org/apache/iceberg/index/TrackingFileReader.java b/core/src/main/java/org/apache/iceberg/index/TrackingFileReader.java new file mode 100644 index 000000000000..85c0a45133ac --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/TrackingFileReader.java @@ -0,0 +1,115 @@ +/* + * 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.io.IOException; +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.util.List; +import org.apache.avro.file.DataFileStream; +import org.apache.avro.generic.GenericDatumReader; +import org.apache.avro.generic.GenericRecord; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; + +/** + * Reads a tracking file written by {@link TrackingFileWriter}. + * + *

Uses Avro's {@link DataFileStream} for sequential streaming reads from Iceberg's + * {@link InputFile} — no random access required, which keeps compatibility with object stores + * where seeking is expensive. + */ +public class TrackingFileReader { + + private TrackingFileReader() {} + + /** + * Read all entries from a tracking file. + * + * @param inputFile the tracking file to read + * @return list of all tracking file entries + */ + public static List readAll(InputFile inputFile) { + Preconditions.checkNotNull(inputFile, "inputFile is required"); + List entries = Lists.newArrayList(); + try (DataFileStream stream = + new DataFileStream<>( + inputFile.newStream(), new GenericDatumReader<>(TrackingFileWriter.AVRO_SCHEMA))) { + while (stream.hasNext()) { + entries.add(fromRecord(stream.next())); + } + } catch (IOException e) { + throw new UncheckedIOException( + "Failed to read tracking file: " + inputFile.location(), e); + } + return entries; + } + + /** + * Read only the entries whose transform-value range overlaps with {@code [queryMin, queryMax]}. + * + *

This is the primary planning-time operation: given a predicate that maps to a transform + * value range, return only the leaf files that could contain matching entries. + * + * @param inputFile the tracking file to read + * @param queryMin minimum transform value for the query (inclusive) + * @param queryMax maximum transform value for the query (inclusive) + * @return list of entries whose transform-value range overlaps the query range + */ + public static List readMatching( + InputFile inputFile, long queryMin, long queryMax) { + Preconditions.checkNotNull(inputFile, "inputFile is required"); + List matches = Lists.newArrayList(); + try (DataFileStream stream = + new DataFileStream<>( + inputFile.newStream(), new GenericDatumReader<>(TrackingFileWriter.AVRO_SCHEMA))) { + while (stream.hasNext()) { + TrackingFileEntry entry = fromRecord(stream.next()); + // Overlap check: entry range [lower, upper] overlaps [queryMin, queryMax] + // iff lower <= queryMax AND upper >= queryMin + if (entry.transformValueLowerBound() <= queryMax + && entry.transformValueUpperBound() >= queryMin) { + matches.add(entry); + } + } + } catch (IOException e) { + throw new UncheckedIOException( + "Failed to read tracking file: " + inputFile.location(), e); + } + return matches; + } + + private static TrackingFileEntry fromRecord(GenericRecord record) { + TrackingFileEntry.Builder builder = + TrackingFileEntry.builder() + .location(record.get("location").toString()) + .fileFormat(record.get("file_format").toString()) + .recordCount((Long) record.get("record_count")) + .fileSizeInBytes((Long) record.get("file_size_in_bytes")) + .transformValueLowerBound((Long) record.get("transform_value_lower_bound")) + .transformValueUpperBound((Long) record.get("transform_value_upper_bound")); + + Object keyMeta = record.get("key_metadata"); + if (keyMeta != null) { + builder.keyMetadata((ByteBuffer) keyMeta); + } + return builder.build(); + } +} diff --git a/core/src/main/java/org/apache/iceberg/index/TrackingFileWriter.java b/core/src/main/java/org/apache/iceberg/index/TrackingFileWriter.java new file mode 100644 index 000000000000..3c4cbf3f33fc --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/TrackingFileWriter.java @@ -0,0 +1,120 @@ +/* + * 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.io.IOException; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.util.List; +import org.apache.avro.Schema; +import org.apache.avro.file.CodecFactory; +import org.apache.avro.file.DataFileWriter; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericDatumWriter; +import org.apache.avro.generic.GenericRecord; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Writes a tracking file in Avro format. + * + *

The tracking file is a metadata file (like an Iceberg manifest) that lists all leaf files + * belonging to an index snapshot, with transform-value bounds per leaf file used for planning-time + * pruning. + * + *

Uses Avro's {@link DataFileWriter} writing to Iceberg's {@link OutputFile}, following the + * same pattern as Iceberg's own manifest writers. + */ +public class TrackingFileWriter implements AutoCloseable { + + static final Schema AVRO_SCHEMA = + new Schema.Parser() + .parse( + "{" + + "\"type\":\"record\"," + + "\"name\":\"tracking_file_entry\"," + + "\"namespace\":\"org.apache.iceberg.index\"," + + "\"fields\":[" + + " {\"name\":\"location\",\"type\":\"string\",\"field-id\":100}," + + " {\"name\":\"file_format\",\"type\":\"string\",\"field-id\":101}," + + " {\"name\":\"record_count\",\"type\":\"long\",\"field-id\":103}," + + " {\"name\":\"file_size_in_bytes\",\"type\":\"long\",\"field-id\":104}," + + " {\"name\":\"transform_value_lower_bound\",\"type\":\"long\",\"field-id\":200}," + + " {\"name\":\"transform_value_upper_bound\",\"type\":\"long\",\"field-id\":201}," + + " {\"name\":\"key_metadata\",\"type\":[\"null\",\"bytes\"],\"default\":null,\"field-id\":131}" + + "]" + + "}"); + + private final DataFileWriter writer; + private final OutputStream stream; + private int entryCount = 0; + + public TrackingFileWriter(OutputFile outputFile) { + Preconditions.checkNotNull(outputFile, "outputFile is required"); + try { + this.stream = outputFile.createOrOverwrite(); + DataFileWriter dfw = + new DataFileWriter(new GenericDatumWriter(AVRO_SCHEMA)) + .setCodec(CodecFactory.snappyCodec()); + dfw.create(AVRO_SCHEMA, stream); + this.writer = dfw; + } catch (IOException e) { + throw new UncheckedIOException("Failed to create tracking file writer", e); + } + } + + /** Write a single tracking file entry. */ + public void add(TrackingFileEntry entry) { + Preconditions.checkNotNull(entry, "entry is required"); + GenericRecord record = new GenericData.Record(AVRO_SCHEMA); + record.put("location", entry.location()); + record.put("file_format", entry.fileFormat()); + record.put("record_count", entry.recordCount()); + record.put("file_size_in_bytes", entry.fileSizeInBytes()); + record.put("transform_value_lower_bound", entry.transformValueLowerBound()); + record.put("transform_value_upper_bound", entry.transformValueUpperBound()); + record.put( + "key_metadata", + entry.keyMetadata() != null ? entry.keyMetadata() : null); + try { + writer.append(record); + entryCount++; + } catch (IOException e) { + throw new UncheckedIOException("Failed to write tracking file entry", e); + } + } + + /** Write all entries from a list. */ + public void addAll(List entries) { + entries.forEach(this::add); + } + + public int entryCount() { + return entryCount; + } + + @Override + public void close() { + try { + writer.close(); + } catch (IOException e) { + throw new UncheckedIOException("Failed to close tracking file writer", e); + } + } +} diff --git a/core/src/test/java/org/apache/iceberg/index/TestTrackingFile.java b/core/src/test/java/org/apache/iceberg/index/TestTrackingFile.java new file mode 100644 index 000000000000..b066c6db5bf6 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/index/TestTrackingFile.java @@ -0,0 +1,247 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThat; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.List; +import org.apache.avro.file.DataFileStream; +import org.apache.avro.file.DataFileWriter; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericDatumReader; +import org.apache.avro.generic.GenericDatumWriter; +import org.apache.avro.generic.GenericRecord; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.PositionOutputStream; +import org.apache.iceberg.io.SeekableInputStream; +import org.junit.jupiter.api.Test; + +public class TestTrackingFile { + + /** In-memory OutputFile backed by a ByteArrayOutputStream. */ + private static class InMemoryOutputFile implements OutputFile { + final ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + private final String location; + + InMemoryOutputFile(String location) { + this.location = location; + } + + @Override + public PositionOutputStream create() { + return new PositionOutputStream() { + private long pos = 0; + + @Override + public long getPos() { + return pos; + } + + @Override + public void write(int b) throws IOException { + buffer.write(b); + pos++; + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + buffer.write(b, off, len); + pos += len; + } + }; + } + + @Override + public PositionOutputStream createOrOverwrite() { + buffer.reset(); + return create(); + } + + @Override + public String location() { + return location; + } + + @Override + public InputFile toInputFile() { + return new InMemoryInputFile(location, buffer.toByteArray()); + } + } + + /** In-memory InputFile backed by a byte array. */ + private static class InMemoryInputFile implements InputFile { + private final String location; + private final byte[] data; + + InMemoryInputFile(String location, byte[] data) { + this.location = location; + this.data = data; + } + + @Override + public long getLength() { + return data.length; + } + + @Override + public SeekableInputStream newStream() { + return new SeekableInputStream() { + private final InputStream delegate = new ByteArrayInputStream(data); + private long pos = 0; + + @Override + public long getPos() { + return pos; + } + + @Override + public void seek(long newPos) throws IOException { + throw new UnsupportedOperationException("seek not supported in test"); + } + + @Override + public int read() throws IOException { + int b = delegate.read(); + if (b >= 0) pos++; + return b; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + int n = delegate.read(b, off, len); + if (n > 0) pos += n; + return n; + } + }; + } + + @Override + public String location() { + return location; + } + + @Override + public boolean exists() { + return true; + } + } + + private List writeAndRead(List entries) { + InMemoryOutputFile outputFile = new InMemoryOutputFile("test://tracking.avro"); + try (TrackingFileWriter writer = new TrackingFileWriter(outputFile)) { + writer.addAll(entries); + } + return TrackingFileReader.readAll(outputFile.toInputFile()); + } + + @Test + void roundTripSingleEntry() { + TrackingFileEntry entry = + TrackingFileEntry.builder() + .location("s3://warehouse/db/orders/index/leaf-00001.parquet") + .fileFormat("parquet") + .recordCount(1000L) + .fileSizeInBytes(204800L) + .transformValueLowerBound(0L) + .transformValueUpperBound(63L) + .build(); + + List restored = writeAndRead(List.of(entry)); + + assertThat(restored).hasSize(1); + TrackingFileEntry r = restored.get(0); + assertThat(r.location()).isEqualTo(entry.location()); + assertThat(r.fileFormat()).isEqualTo("parquet"); + assertThat(r.recordCount()).isEqualTo(1000L); + assertThat(r.fileSizeInBytes()).isEqualTo(204800L); + assertThat(r.transformValueLowerBound()).isEqualTo(0L); + assertThat(r.transformValueUpperBound()).isEqualTo(63L); + assertThat(r.keyMetadata()).isNull(); + } + + @Test + void roundTripMultipleEntries() { + List entries = List.of( + TrackingFileEntry.builder() + .location("s3://.../leaf-0.parquet").recordCount(500) + .fileSizeInBytes(1024).transformValueLowerBound(0).transformValueUpperBound(63).build(), + TrackingFileEntry.builder() + .location("s3://.../leaf-1.parquet").recordCount(600) + .fileSizeInBytes(2048).transformValueLowerBound(64).transformValueUpperBound(127).build(), + TrackingFileEntry.builder() + .location("s3://.../leaf-2.parquet").recordCount(700) + .fileSizeInBytes(3072).transformValueLowerBound(128).transformValueUpperBound(255).build() + ); + + List restored = writeAndRead(entries); + assertThat(restored).hasSize(3); + assertThat(restored.get(0).transformValueLowerBound()).isEqualTo(0L); + assertThat(restored.get(1).transformValueLowerBound()).isEqualTo(64L); + assertThat(restored.get(2).transformValueUpperBound()).isEqualTo(255L); + } + + @Test + void readMatchingFiltersCorrectly() { + List entries = List.of( + TrackingFileEntry.builder().location("leaf-0.parquet").recordCount(100) + .fileSizeInBytes(1024).transformValueLowerBound(0).transformValueUpperBound(63).build(), + TrackingFileEntry.builder().location("leaf-1.parquet").recordCount(100) + .fileSizeInBytes(1024).transformValueLowerBound(64).transformValueUpperBound(127).build(), + TrackingFileEntry.builder().location("leaf-2.parquet").recordCount(100) + .fileSizeInBytes(1024).transformValueLowerBound(128).transformValueUpperBound(255).build() + ); + + InMemoryOutputFile outputFile = new InMemoryOutputFile("test://tracking.avro"); + try (TrackingFileWriter writer = new TrackingFileWriter(outputFile)) { + writer.addAll(entries); + } + InputFile inputFile = outputFile.toInputFile(); + + // Query for bucket 42 — only leaf-0 (0..63) should match + List matches = TrackingFileReader.readMatching(inputFile, 42, 42); + assertThat(matches).hasSize(1); + assertThat(matches.get(0).location()).isEqualTo("leaf-0.parquet"); + + // Query for range 60..70 — leaf-0 (0..63) and leaf-1 (64..127) both overlap + List rangeMatches = TrackingFileReader.readMatching(inputFile, 60, 70); + assertThat(rangeMatches).hasSize(2); + + // Query for bucket 300 — no match + List noMatches = TrackingFileReader.readMatching(inputFile, 300, 300); + assertThat(noMatches).isEmpty(); + } + + @Test + void writerCountsEntries() { + InMemoryOutputFile outputFile = new InMemoryOutputFile("test://tracking.avro"); + try (TrackingFileWriter writer = new TrackingFileWriter(outputFile)) { + writer.add(TrackingFileEntry.builder().location("a.parquet").recordCount(1) + .fileSizeInBytes(100).transformValueLowerBound(0).transformValueUpperBound(10).build()); + writer.add(TrackingFileEntry.builder().location("b.parquet").recordCount(2) + .fileSizeInBytes(200).transformValueLowerBound(11).transformValueUpperBound(20).build()); + assertThat(writer.entryCount()).isEqualTo(2); + } + } +} From 539b5a1d2af146f058b6b8c9f983ecb22549ee48 Mon Sep 17 00:00:00 2001 From: Manish Malhotra Date: Tue, 21 Jul 2026 04:09:37 -0700 Subject: [PATCH 5/5] Core: Add HashTransform, LeafFileMetadata, ScalarIndexCommitter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the core build logic for the SCALAR index: - HashTransform: maps key values to hash buckets [0, numBuckets). Phase 1 uses Java hashCode; will align with Iceberg murmur3 bucket transform in a follow-up. - LeafFileMetadata: captures path, bounds, record count, and size of one written leaf file; converts to TrackingFileEntry for the tracking file writer. - ScalarIndexCommitter: given a list of leaf files from the Spark build job, writes the tracking file (Avro), creates an IndexSnapshot, writes the metadata JSON, and commits to the IndexCatalog with CAS semantics. Also adds scripts/build_scalar_index_taxi.scala — an end-to-end Spark script demonstrating the full build flow on NYC Yellow Taxi data using the medallion column as the SCALAR HASH index key. 25/25 unit tests passing. --- .../apache/iceberg/index/HashTransform.java | 83 ++++ .../iceberg/index/LeafFileMetadata.java | 96 +++++ .../iceberg/index/ScalarIndexCommitter.java | 173 +++++++++ .../iceberg/index/TestScalarIndexBuild.java | 363 ++++++++++++++++++ scripts/build_scalar_index_taxi.scala | 156 ++++++++ 5 files changed, 871 insertions(+) create mode 100644 core/src/main/java/org/apache/iceberg/index/HashTransform.java create mode 100644 core/src/main/java/org/apache/iceberg/index/LeafFileMetadata.java create mode 100644 core/src/main/java/org/apache/iceberg/index/ScalarIndexCommitter.java create mode 100644 core/src/test/java/org/apache/iceberg/index/TestScalarIndexBuild.java create mode 100644 scripts/build_scalar_index_taxi.scala diff --git a/core/src/main/java/org/apache/iceberg/index/HashTransform.java b/core/src/main/java/org/apache/iceberg/index/HashTransform.java new file mode 100644 index 000000000000..8f575c1b4cef --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/HashTransform.java @@ -0,0 +1,83 @@ +/* + * 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 org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * HASH transform for the SCALAR index type. + * + *

Maps a key value to a hash bucket in [0, numBuckets). Uses the same + * murmur3-based hash as Iceberg's bucket partition transform so that index + * bucket assignments are consistent with partition pruning. + * + *

The transform value stored in leaf files is the bucket number (long). + * The tracking file stores [bucketMin, bucketMax] per leaf file, enabling + * the planner to identify which leaf files to scan for a given key. + */ +public class HashTransform { + + private final int numBuckets; + + public HashTransform(int numBuckets) { + Preconditions.checkArgument(numBuckets > 0, "numBuckets must be positive, got: %s", numBuckets); + this.numBuckets = numBuckets; + } + + public int numBuckets() { + return numBuckets; + } + + /** + * Compute the hash bucket for a String value. + * + *

Uses Java's {@code hashCode()} for Phase 1. A follow-up will switch to murmur3_x86_32 + * to align with Iceberg's bucket partition transform. + */ + public long apply(String value) { + if (value == null) { + return 0; + } + return Math.floorMod(value.hashCode(), numBuckets); + } + + /** Compute the hash bucket for a long value. */ + public long apply(long value) { + return Math.floorMod(Long.hashCode(value), numBuckets); + } + + /** Compute the hash bucket for an int value. */ + public long apply(int value) { + return Math.floorMod(Integer.hashCode(value), numBuckets); + } + + /** + * Compute the hash bucket for a value of any supported type. + * + * @throws IllegalArgumentException if the type is not supported + */ + public long apply(Object value) { + if (value == null) return 0; + if (value instanceof String) return apply((String) value); + if (value instanceof Long) return apply((long) (Long) value); + if (value instanceof Integer) return apply((int) (Integer) value); + throw new IllegalArgumentException( + "Unsupported type for HASH transform: " + value.getClass().getName()); + } +} diff --git a/core/src/main/java/org/apache/iceberg/index/LeafFileMetadata.java b/core/src/main/java/org/apache/iceberg/index/LeafFileMetadata.java new file mode 100644 index 000000000000..9046e51df465 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/LeafFileMetadata.java @@ -0,0 +1,96 @@ +/* + * 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 org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Metadata about a leaf file written during index build. + * + *

Captures the path, size, record count, and transform-value bounds of + * one leaf file. These fields are written to the tracking file so the planner + * can select the right leaf files at planning time without opening them. + */ +public class LeafFileMetadata { + + private final String path; + private final String fileFormat; + private final long recordCount; + private final long fileSizeInBytes; + private final long transformValueMin; + private final long transformValueMax; + + public LeafFileMetadata( + String path, + String fileFormat, + long recordCount, + long fileSizeInBytes, + long transformValueMin, + long transformValueMax) { + Preconditions.checkArgument(path != null && !path.isEmpty(), "path is required"); + Preconditions.checkArgument( + transformValueMin <= transformValueMax, + "transformValueMin (%s) must be <= transformValueMax (%s)", + transformValueMin, transformValueMax); + this.path = path; + this.fileFormat = fileFormat != null ? fileFormat : "parquet"; + this.recordCount = recordCount; + this.fileSizeInBytes = fileSizeInBytes; + this.transformValueMin = transformValueMin; + this.transformValueMax = transformValueMax; + } + + public String path() { + return path; + } + + public String fileFormat() { + return fileFormat; + } + + public long recordCount() { + return recordCount; + } + + public long fileSizeInBytes() { + return fileSizeInBytes; + } + + /** Minimum transform value (e.g. hash bucket) stored in this leaf file. */ + public long transformValueMin() { + return transformValueMin; + } + + /** Maximum transform value (e.g. hash bucket) stored in this leaf file. */ + public long transformValueMax() { + return transformValueMax; + } + + /** Convert to a {@link TrackingFileEntry} for writing to the tracking file. */ + public TrackingFileEntry toTrackingEntry() { + return TrackingFileEntry.builder() + .location(path) + .fileFormat(fileFormat) + .recordCount(recordCount) + .fileSizeInBytes(fileSizeInBytes) + .transformValueLowerBound(transformValueMin) + .transformValueUpperBound(transformValueMax) + .build(); + } +} diff --git a/core/src/main/java/org/apache/iceberg/index/ScalarIndexCommitter.java b/core/src/main/java/org/apache/iceberg/index/ScalarIndexCommitter.java new file mode 100644 index 000000000000..de345fe62221 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/index/ScalarIndexCommitter.java @@ -0,0 +1,173 @@ +/* + * 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 java.util.UUID; +import java.util.concurrent.ThreadLocalRandom; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; + +/** + * Commits a completed index build to the {@link IndexCatalog}. + * + *

Given a list of {@link LeafFileMetadata} produced by the Spark build job, this class: + *

    + *
  1. Writes the tracking file (Avro) to the index metadata directory. + *
  2. Creates a new {@link IndexSnapshot} pointing to the tracking file. + *
  3. Writes the new index metadata JSON file. + *
  4. Atomically registers or updates the index in the {@link IndexCatalog}. + *
+ * + *

The commit is optimistic: if another writer has already committed a newer version, + * {@link java.util.ConcurrentModificationException} is thrown and the caller should retry. + */ +public class ScalarIndexCommitter { + + private final IndexCatalog catalog; + private final FileIO fileIO; + + public ScalarIndexCommitter(IndexCatalog catalog, FileIO fileIO) { + Preconditions.checkNotNull(catalog, "IndexCatalog is required"); + Preconditions.checkNotNull(fileIO, "FileIO is required"); + this.catalog = catalog; + this.fileIO = fileIO; + } + + /** + * Commit a full index build or incremental update. + * + * @param identifier identifies the index in the catalog + * @param tableUuid UUID of the source Iceberg table + * @param sourceTableSnapshotId the table snapshot the index was built from + * @param indexType e.g. {@code "SCALAR"} + * @param transformFunction e.g. {@code "HASH"} or {@code "IDENTITY"} + * @param keyColumnIds source-table column IDs used as index keys + * @param includedColumnIds optional covering column IDs (may be empty) + * @param properties additional index properties (e.g. hash.num-buckets) + * @param indexLocation base location for index files + * @param leafFiles metadata of the leaf files written by the build job + */ + public void commit( + IndexIdentifier identifier, + String tableUuid, + long sourceTableSnapshotId, + String indexType, + String transformFunction, + List keyColumnIds, + List includedColumnIds, + Map properties, + String indexLocation, + List leafFiles) { + + Preconditions.checkArgument( + leafFiles != null && !leafFiles.isEmpty(), "leafFiles must be non-empty"); + + // Load current metadata if the index already exists (incremental update) + IndexMetadata current = null; + if (catalog.indexExists(identifier)) { + current = catalog.loadIndex(identifier); + } + + // 1. Write tracking file (Avro) + String trackingFilePath = newTrackingFilePath(indexLocation, current); + OutputFile trackingOutput = fileIO.newOutputFile(trackingFilePath); + try (TrackingFileWriter writer = new TrackingFileWriter(trackingOutput)) { + for (LeafFileMetadata lf : leafFiles) { + writer.add(lf.toTrackingEntry()); + } + } + + // 2. Build new IndexSnapshot + long newSnapshotId = Math.abs(ThreadLocalRandom.current().nextLong()); + IndexSnapshot snapshot = + GenericIndexSnapshot.builder() + .snapshotId(newSnapshotId) + .sourceTableSnapshotId(sourceTableSnapshotId) + .timestampMs(System.currentTimeMillis()) + .trackingFile(trackingFilePath) + .build(); + + // 3. Build updated IndexMetadata + GenericIndexMetadata.Builder metaBuilder; + if (current == null) { + metaBuilder = + GenericIndexMetadata.builder() + .uuid(UUID.randomUUID().toString()) + .tableUuid(tableUuid) + .location(indexLocation) + .type(indexType) + .transformFunction(transformFunction) + .keyColumnIds(keyColumnIds) + .includedColumnIds(includedColumnIds != null ? includedColumnIds : ImmutableList.of()) + .properties(properties != null ? properties : ImmutableMap.of()); + } else { + metaBuilder = GenericIndexMetadata.buildFrom(current); + } + metaBuilder.addSnapshot(snapshot); + + // 4. Write new metadata file + String newMetadataPath = IndexMetadataIO.newMetadataFileLocation(indexLocation, current); + IndexMetadata updated = metaBuilder.metadataFileLocation(newMetadataPath).build(); + IndexMetadataIO.write(updated, fileIO.newOutputFile(newMetadataPath)); + + // 5. Register or update in catalog (optimistic CAS) + if (current == null) { + catalog.createIndex(identifier, updated); + } else { + catalog.updateIndex(identifier, current, updated); + } + } + + /** Convenience overload without included columns or extra properties. */ + public void commit( + IndexIdentifier identifier, + String tableUuid, + long sourceTableSnapshotId, + String indexType, + String transformFunction, + List keyColumnIds, + String indexLocation, + List leafFiles) { + commit( + identifier, + tableUuid, + sourceTableSnapshotId, + indexType, + transformFunction, + keyColumnIds, + ImmutableList.of(), + ImmutableMap.of(), + indexLocation, + leafFiles); + } + + private static String newTrackingFilePath(String indexLocation, IndexMetadata current) { + int version = current == null ? 1 : current.snapshots().size() + 1; + String suffix = UUID.randomUUID().toString().replace("-", "").substring(0, 8); + return String.format( + "%s/metadata/tracking-%05d-%s.avro", + indexLocation.replaceAll("/$", ""), version, suffix); + } +} diff --git a/core/src/test/java/org/apache/iceberg/index/TestScalarIndexBuild.java b/core/src/test/java/org/apache/iceberg/index/TestScalarIndexBuild.java new file mode 100644 index 000000000000..2c48ef03d62e --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/index/TestScalarIndexBuild.java @@ -0,0 +1,363 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThat; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.ConcurrentModificationException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.PositionOutputStream; +import org.apache.iceberg.io.SeekableInputStream; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class TestScalarIndexBuild { + + private static final TableIdentifier TABLE = + TableIdentifier.of(Namespace.of("taxi"), "yellow_trips"); + private static final IndexIdentifier IDX = + IndexIdentifier.of(TABLE, "medallion_idx"); + private static final String TABLE_UUID = "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94"; + private static final String INDEX_LOCATION = "s3://warehouse/taxi/yellow_trips/index/medallion_idx"; + + private InMemoryIndexCatalog catalog; + private InMemoryFileIO fileIO; + private ScalarIndexCommitter committer; + + @BeforeEach + void setup() { + catalog = new InMemoryIndexCatalog(); + fileIO = new InMemoryFileIO(); + committer = new ScalarIndexCommitter(catalog, fileIO); + } + + // ------------------------------------------------------------------ + // HashTransform tests + // ------------------------------------------------------------------ + + @Test + void hashTransformStringDeterministic() { + HashTransform t = new HashTransform(256); + // Same value always produces same bucket + assertThat(t.apply("D7D598CD99978BD012A87A76A7C891B7")) + .isEqualTo(t.apply("D7D598CD99978BD012A87A76A7C891B7")); + } + + @Test + void hashTransformBucketsInRange() { + HashTransform t = new HashTransform(256); + for (String val : List.of("medallion1", "medallion2", "hello", "world", "abc123")) { + long bucket = t.apply(val); + assertThat(bucket).isBetween(0L, 255L); + } + } + + @Test + void hashTransformDistributesEvenly() { + HashTransform t = new HashTransform(256); + long[] counts = new long[256]; + for (int i = 0; i < 10_000; i++) { + counts[(int) t.apply("medallion_" + i)]++; + } + // Each bucket should have roughly 10000/256 ≈ 39 entries + // Check no bucket has more than 3x the average (basic sanity) + long avg = 10_000 / 256; + for (long count : counts) { + assertThat(count).isLessThan(avg * 5); + } + } + + // ------------------------------------------------------------------ + // Full build + commit cycle + // ------------------------------------------------------------------ + + @Test + void firstBuildCreatesIndexInCatalog() { + List leafFiles = sampleLeafFiles(); + + committer.commit( + IDX, TABLE_UUID, 3055729675574597004L, + "SCALAR", "HASH", + ImmutableList.of(3), + ImmutableList.of(), + ImmutableMap.of("hash.num-buckets", "256"), + INDEX_LOCATION, + leafFiles); + + assertThat(catalog.indexExists(IDX)).isTrue(); + IndexMetadata loaded = catalog.loadIndex(IDX); + assertThat(loaded.type()).isEqualTo("SCALAR"); + assertThat(loaded.transformFunction()).isEqualTo("HASH"); + assertThat(loaded.snapshots()).hasSize(1); + assertThat(loaded.currentSnapshotId()).isNotNull(); + + // Verify the tracking file was written + IndexSnapshot snap = loaded.currentSnapshot(); + assertThat(snap.sourceTableSnapshotId()).isEqualTo(3055729675574597004L); + assertThat(snap.trackingFile()).startsWith(INDEX_LOCATION + "/metadata/tracking-00001-"); + assertThat(fileIO.files).containsKey(snap.trackingFile()); + } + + @Test + void incrementalBuildAddsSecondSnapshot() { + // First build + committer.commit( + IDX, TABLE_UUID, 1000L, + "SCALAR", "HASH", ImmutableList.of(3), + INDEX_LOCATION, sampleLeafFiles()); + + // Second build (new table snapshot) + committer.commit( + IDX, TABLE_UUID, 2000L, + "SCALAR", "HASH", ImmutableList.of(3), + INDEX_LOCATION, sampleLeafFiles()); + + IndexMetadata loaded = catalog.loadIndex(IDX); + assertThat(loaded.snapshots()).hasSize(2); + assertThat(loaded.snapshotForTableSnapshot(1000L)).isNotNull(); + assertThat(loaded.snapshotForTableSnapshot(2000L)).isNotNull(); + + // Latest snapshot should be current + assertThat(loaded.currentSnapshot().sourceTableSnapshotId()).isEqualTo(2000L); + } + + @Test + void trackingFileContainsCorrectBounds() { + List leafFiles = List.of( + new LeafFileMetadata("s3://.../leaf-0.parquet", "parquet", 1000, 204800, 0, 63), + new LeafFileMetadata("s3://.../leaf-1.parquet", "parquet", 900, 192000, 64, 127), + new LeafFileMetadata("s3://.../leaf-2.parquet", "parquet", 1100, 220000, 128, 255) + ); + + committer.commit( + IDX, TABLE_UUID, 1000L, + "SCALAR", "HASH", ImmutableList.of(3), + INDEX_LOCATION, leafFiles); + + // Read back the tracking file + String trackingPath = catalog.loadIndex(IDX).currentSnapshot().trackingFile(); + InputFile trackingFile = fileIO.newInputFile(trackingPath); + List entries = TrackingFileReader.readAll(trackingFile); + + assertThat(entries).hasSize(3); + assertThat(entries.get(0).transformValueLowerBound()).isEqualTo(0L); + assertThat(entries.get(0).transformValueUpperBound()).isEqualTo(63L); + assertThat(entries.get(2).transformValueUpperBound()).isEqualTo(255L); + } + + @Test + void plannerFindsRightLeafFileForMedallion() { + HashTransform transform = new HashTransform(256); + String medallion = "D7D598CD99978BD012A87A76A7C891B7"; + long bucket = transform.apply(medallion); + + // Simulate 4 leaf files covering 64 buckets each + List leafFiles = List.of( + new LeafFileMetadata("s3://.../leaf-0.parquet", "parquet", 500, 100000, 0, 63), + new LeafFileMetadata("s3://.../leaf-1.parquet", "parquet", 500, 100000, 64, 127), + new LeafFileMetadata("s3://.../leaf-2.parquet", "parquet", 500, 100000, 128, 191), + new LeafFileMetadata("s3://.../leaf-3.parquet", "parquet", 500, 100000, 192, 255) + ); + + committer.commit( + IDX, TABLE_UUID, 1000L, "SCALAR", "HASH", + ImmutableList.of(3), INDEX_LOCATION, leafFiles); + + // Simulate planning: use readMatching to find the right leaf file + String trackingPath = catalog.loadIndex(IDX).currentSnapshot().trackingFile(); + InputFile trackingFile = fileIO.newInputFile(trackingPath); + List matches = TrackingFileReader.readMatching(trackingFile, bucket, bucket); + + // Should narrow down to exactly 1 leaf file + assertThat(matches).hasSize(1); + assertThat(matches.get(0).transformValueLowerBound()).isLessThanOrEqualTo(bucket); + assertThat(matches.get(0).transformValueUpperBound()).isGreaterThanOrEqualTo(bucket); + } + + // ------------------------------------------------------------------ + // Helpers + // ------------------------------------------------------------------ + + private List sampleLeafFiles() { + return List.of( + new LeafFileMetadata(INDEX_LOCATION + "/data/leaf-0.parquet", "parquet", 1000, 204800, 0, 127), + new LeafFileMetadata(INDEX_LOCATION + "/data/leaf-1.parquet", "parquet", 1100, 225000, 128, 255) + ); + } + + /** + * In-memory FileIO backed by a HashMap of path → byte[]. + * Allows tests to read back files written by the committer. + */ + static class InMemoryFileIO implements FileIO { + final Map files = new HashMap<>(); + + @Override + public InputFile newInputFile(String path) { + byte[] data = files.get(path); + if (data == null) throw new RuntimeException("File not found: " + path); + return new InMemInput(path, data); + } + + @Override + public InputFile newInputFile(String path, long length) { + return newInputFile(path); + } + + @Override + public OutputFile newOutputFile(String path) { + return new InMemOutput(path, files); + } + + @Override + public void deleteFile(String path) { + files.remove(path); + } + } + + private static class InMemOutput implements OutputFile { + private final String path; + private final Map store; + private final ByteArrayOutputStream buf = new ByteArrayOutputStream(); + + InMemOutput(String path, Map store) { + this.path = path; + this.store = store; + } + + @Override + public PositionOutputStream create() { + return pos(); + } + + @Override + public PositionOutputStream createOrOverwrite() { + buf.reset(); + return pos(); + } + + private PositionOutputStream pos() { + return new PositionOutputStream() { + private long p = 0; + + @Override + public long getPos() { + return p; + } + + @Override + public void write(int b) throws IOException { + buf.write(b); + p++; + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + buf.write(b, off, len); + p += len; + } + + @Override + public void close() throws IOException { + super.close(); + store.put(path, buf.toByteArray()); + } + }; + } + + @Override + public String location() { + return path; + } + + @Override + public InputFile toInputFile() { + return new InMemInput(path, buf.toByteArray()); + } + } + + private static class InMemInput implements InputFile { + private final String path; + private final byte[] data; + + InMemInput(String path, byte[] data) { + this.path = path; + this.data = data; + } + + @Override + public long getLength() { + return data.length; + } + + @Override + public SeekableInputStream newStream() { + return new SeekableInputStream() { + private final ByteArrayInputStream in = new ByteArrayInputStream(data); + private long pos = 0; + + @Override + public long getPos() { + return pos; + } + + @Override + public void seek(long newPos) { + throw new UnsupportedOperationException(); + } + + @Override + public int read() throws IOException { + int b = in.read(); + if (b >= 0) pos++; + return b; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + int n = in.read(b, off, len); + if (n > 0) pos += n; + return n; + } + }; + } + + @Override + public String location() { + return path; + } + + @Override + public boolean exists() { + return true; + } + } +} diff --git a/scripts/build_scalar_index_taxi.scala b/scripts/build_scalar_index_taxi.scala new file mode 100644 index 000000000000..29d804cf3d4b --- /dev/null +++ b/scripts/build_scalar_index_taxi.scala @@ -0,0 +1,156 @@ +/** + * Build a SCALAR HASH index on NYC Yellow Taxi data (medallion column). + * + * Run with spark-shell or spark-submit: + * spark-shell --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.10.0 \ + * --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \ + * --conf spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog \ + * --conf spark.sql.catalog.local.type=hadoop \ + * --conf spark.sql.catalog.local.warehouse=/tmp/iceberg-warehouse \ + * -i scripts/build_scalar_index_taxi.scala + * + * Or paste directly into spark-shell. + */ + +import org.apache.spark.sql.functions._ +import org.apache.spark.sql.expressions.Window +import org.apache.iceberg.index._ +import org.apache.iceberg.catalog.{TableIdentifier, Namespace} +import org.apache.iceberg.hadoop.HadoopFileIO +import com.google.common.collect.ImmutableList +import java.io.File + +// ── 1. Setup ────────────────────────────────────────────────────────────────── + +val TABLE_NAME = "local.taxi.yellow_trips" +val INDEX_LOCATION = "/tmp/iceberg-warehouse/taxi/yellow_trips/index/medallion_idx" +val NUM_BUCKETS = 256 +val KEY_COLUMN = "medallion" +val KEY_COLUMN_ID = 3 // Iceberg field id for medallion in the table schema + +// ── 2. Load source table ─────────────────────────────────────────────────────── + +println(s"Reading $TABLE_NAME ...") +val taxiDf = spark.read.format("iceberg").load(TABLE_NAME) +println(s"Total rows: ${taxiDf.count()}") +println(s"Source files: approximately ${taxiDf.select(input_file_name()).distinct().count()}") + +// ── 3. Compute transform values and collect file metadata ───────────────────── + +val transform = new HashTransform(NUM_BUCKETS) + +// Broadcast the transform to executors +val numBuckets = NUM_BUCKETS +val withTransform = taxiDf + .select( + col(KEY_COLUMN), + input_file_name().as("source_file_path") + ) + // Compute hash bucket on the driver-broadcast numBuckets + .withColumn( + "transform_value", + (hash(col(KEY_COLUMN)) % numBuckets + numBuckets) % numBuckets + ) + +println("Sample transform values:") +withTransform.show(5) + +// ── 4. Write sorted leaf files (Parquet) ────────────────────────────────────── + +val leafOutputPath = s"$INDEX_LOCATION/data" +println(s"Writing leaf files to $leafOutputPath ...") + +withTransform + .repartitionByRange(numBuckets / 64, col("transform_value")) // ~4 leaf files + .sortWithinPartitions(col("transform_value"), col(KEY_COLUMN)) + .write + .format("parquet") + .mode("overwrite") + .save(leafOutputPath) + +println("Leaf files written.") + +// ── 5. Collect leaf file metadata (path, count, size, bounds) ───────────────── + +val leafFiles = spark.read.parquet(leafOutputPath) + .select( + input_file_name().as("path"), + col("transform_value") + ) + .groupBy("path") + .agg( + count("*").as("record_count"), + min("transform_value").as("tv_min"), + max("transform_value").as("tv_max") + ) + .collect() + .map { row => + val path = row.getString(0) + val count = row.getLong(1) + val tvMin = row.getLong(2) + val tvMax = row.getLong(3) + val sizeBytes = new File(path.replace("file:", "")).length() + new LeafFileMetadata(path, "parquet", count, sizeBytes, tvMin, tvMax) + } + .toList + +println(s"Leaf file count: ${leafFiles.size}") +leafFiles.foreach { lf => + println(s" ${lf.path().split("/").last} | rows=${lf.recordCount()} " + + s"| buckets=[${lf.transformValueMin()}, ${lf.transformValueMax()}]") +} + +// ── 6. Commit index via ScalarIndexCommitter ─────────────────────────────────── + +val hadoopConf = spark.sparkContext.hadoopConfiguration +val fileIO = new HadoopFileIO(hadoopConf) +val catalog = new InMemoryIndexCatalog() // swap for HadoopIndexCatalog in production +val committer = new ScalarIndexCommitter(catalog, fileIO) + +val tableIdent = TableIdentifier.of(Namespace.of("taxi"), "yellow_trips") +val indexIdent = IndexIdentifier.of(tableIdent, "medallion_idx") + +// Get the current table snapshot id +val icebergTable = spark.sessionState.catalogManager + .catalog("local") + .asInstanceOf[org.apache.iceberg.spark.SparkCatalog] + .loadTable(tableIdent.asInstanceOf[org.apache.iceberg.catalog.TableIdentifier]) +val tableSnapshotId = icebergTable.currentSnapshot().snapshotId() + +import scala.jdk.CollectionConverters._ +committer.commit( + indexIdent, + icebergTable.uuid(), + tableSnapshotId, + "SCALAR", + "HASH", + ImmutableList.of(KEY_COLUMN_ID), + ImmutableList.of(), + Map("hash.num-buckets" -> NUM_BUCKETS.toString).asJava, + INDEX_LOCATION, + leafFiles.asJava +) + +println(s"\n✅ Index committed: $indexIdent") +val meta = catalog.loadIndex(indexIdent) +println(s" UUID: ${meta.uuid()}") +println(s" Snapshot: ${meta.currentSnapshotId()}") +println(s" Tracking: ${meta.currentSnapshot().trackingFile()}") + +// ── 7. Simulate a planner lookup ────────────────────────────────────────────── + +val queryMedallion = "D7D598CD99978BD012A87A76A7C891B7" +val queryBucket = transform.apply(queryMedallion) +println(s"\n🔍 Planning query: WHERE medallion = '$queryMedallion'") +println(s" Hash bucket: $queryBucket") + +val trackingPath = meta.currentSnapshot().trackingFile() +val trackingFile = fileIO.newInputFile(trackingPath) +val matchingLeafFiles = TrackingFileReader.readMatching(trackingFile, queryBucket, queryBucket) + +println(s" Leaf files to scan: ${matchingLeafFiles.size()} (out of ${leafFiles.size} total)") +matchingLeafFiles.forEach { entry => + println(s" → ${entry.location().split("/").last} | buckets=[${entry.transformValueLowerBound()},${entry.transformValueUpperBound()}]") +} +println(s"\n Without index: scan all ${taxiDf.select(input_file_name()).distinct().count()} source files") +println(s" With index: scan 1 leaf file → at most 1 source file")