Conversation
Applies the index spec document from apache/iceberg PR apache#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.
Implements IndexMetadata and IndexSnapshot POJOs matching the field definitions in format/index.md (PR apache#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.
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.
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.
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.
…hot, and commit validation Fixes three errorprone ERROR-level findings that blocked compilation: missing Locale.ROOT in String.format calls in ScalarIndexCommitter and IndexMetadataIO, and a RuntimeException that should have been UncheckedIOException in IndexSnapshotParser. Adds test coverage that was previously missing entirely: - IndexMetadataIO.read() round-trip against the actual bytes written by ScalarIndexCommitter, not just the in-memory catalog object. - GenericIndexMetadata.removeSnapshot(), including currentSnapshotId reassignment when the removed snapshot was current. - ScalarIndexCommitter validation for null/empty leafFiles. - A deterministic characterization test (via a new RaceInjectingCatalog test double) pinning down that a race on the first commit for an identifier throws AlreadyExistsException, not the ConcurrentModificationException the class's javadoc says callers should retry on. This documents current behavior; it is not a fix.
…dex reads Implements the read side of the SCALAR index, which previously did not exist at all -- only the build/commit side (ScalarIndexCommitter, TrackingFileWriter/Reader) was implemented before this. Lives in iceberg-data, not iceberg-core, since leaf files are Parquet and iceberg-core's main sourceset has no compile-time dependency on iceberg-parquet (only testRuntimeOnly) -- that boundary looks intentional (core is kept format-agnostic), so this follows the project's existing convention of putting Parquet-specific I/O where the dependency already exists, same as GenericParquetWriter/Readers. LeafFileEntry.schema(NestedField keyField) is the single source of truth for the leaf-file schema, used by both writer and reader, so they cannot drift apart on field IDs. The three synthetic fields (transform_value, file_path, position) use Integer.MAX_VALUE-(101-103) as field IDs, placed just past Iceberg's own reserved metadata-column range so they never collide with a source table's real field IDs or Iceberg's own metadata columns. LeafFileReader.readMatching pushes the predicate to Parquet via ReadBuilder#filter for row-group-level statistics pruning, then applies an Evaluator bound to the leaf schema for exact per-record matching. This second step is necessary: the createReaderFunc-based read path (org.apache.iceberg.parquet.ParquetReader) only skips whole row groups by statistics and does not filter individual records within a row group that wasn't skipped, unlike the older readSupport-based path. Confirmed by two failing tests before this fix -- filtered reads were returning every row in the file. LeafFileWriter does not sort; it writes rows in whatever order the caller provides. Correct pruning depends on the caller (the future index build job) having sorted by (transform_value, key_value) first -- this is not yet enforced or verified anywhere.
LeafFileWriter now tracks the last written (transform_value, key_value) tuple and throws IllegalStateException immediately if a new entry is out of order, rather than silently writing an unsorted leaf file. This matters even for today's single-column HASH/IDENTITY transforms, not just a hypothetical multi-column future: unsorted leaf-file rows mean Parquet's row-group statistics can't prune anything, since every row group ends up spanning close to the full transform_value range. Results stay correct either way (row-group skip is always a safe exclusion, and the per-record Evaluator in LeafFileReader guarantees exact matches), but pruning effectiveness collapses silently -- no error, no warning, just a query that's slow for no visible reason. Deliberately does not sort internally (that's the rejected alternative, option 1): the real build job already has to shuffle/partition data by transform_value bucket to write leaf files at all, and Spark's sortWithinPartitions gets the sort essentially for free as part of that same operation. Sorting again here would duplicate that cost on every leaf file to guard against a bug a correct caller never triggers. Equal consecutive keys are allowed, since a key value is not required to be unique across rows.
Adds TestScalarIndexEndToEnd, exercising the full SCALAR index chain as one flow for the first time: write sorted leaf files via LeafFileWriter, commit them through ScalarIndexCommitter into an InMemoryIndexCatalog, then look up a key exactly the way a planner would -- compute its transform value via HashTransform, narrow to candidate leaf files via TrackingFileReader, then resolve the exact row within those leaf files via LeafFileReader. Covers both an exact-match lookup (verifying the correct file_path/position comes back) and a missing-key lookup (verifying an empty result). Until now the build/commit side and the read side were two separately- tested halves that had never actually been exercised together. This doesn't wire the index into any real engine or source table -- it proves the pieces we've built work together correctly, nothing more. Uses a minimal local-filesystem FileIO test double (real files, since Parquet needs seekable I/O) rather than the in-memory doubles used elsewhere for Avro-only tests.
Implements "CALL system.build_scalar_index(table, columns, transform, options)", populating a SCALAR index from an actual Spark job for the first time -- everything before this was either a standalone demo script or a JUnit test manually orchestrating the pieces. Computes each row's position via row_number() over a window partitioned by input_file_name(), before any shuffle, so it reflects physical file-scan order. Computes transform_value by calling the real HashTransform class through a typed UDF (HASH) or a direct cast (IDENTITY, numeric keys only, since a string can't cast to the long transform_value column) -- never reimplements the transform logic in Spark-land. Shuffles by transform_value bucket and sorts within partitions, then writes one leaf file per partition via LeafFileWriter, collecting per-partition metadata back to the driver via Encoders.javaSerialization (a Dataset<Row> encoder for a custom struct isn't a stable enough API to rely on here). Commits through ScalarIndexCommitter into a new session-scoped registry, SparkIndexCatalogs, added alongside this and following the same singleton pattern as the existing ScanTaskSetManager. Only a single key column is supported (multi-column composite indexes remain an explicit Non-Goal in the design proposal); the procedure rejects multiple columns with a clear error rather than silently using only the first one. Important caveat: this Spark code has not been compiled or run. Gradle cannot even start in this sandbox (blocked at the local socket/daemon level), so every Iceberg-core API call was verified against the actual source in this repo, but the Spark-specific pieces (UDF registration, window functions, mapPartitions encoder usage) rely on general Spark API knowledge plus in-repo precedent where found, not execution. Needs a real compile and test run before being trusted.
Adds tryPruneUsingScalarIndex(), called from pushPredicates() after normal predicate pushdown. On an equality predicate matching an indexed column, resolves the exact source file via TrackingFileReader + LeafFileReader (using SparkIndexCatalogs to find the index) and, if it resolves to exactly one file, adds an Expressions.equal(_file, ...) constraint on top of the existing filters. File-level pruning only in this pass -- true row-position pushdown into the scan tasks is a further, not-yet-attempted refinement, noted in the method's javadoc. Deliberately fail-open everywhere: no index registered, a stale index snapshot relative to the table's current snapshot, an unsupported predicate shape, zero or multiple leaf-file matches, or any I/O error all fall back silently to normal planning. For a table with no index (the common case) this method is a complete no-op, which is the whole point -- existing tables and existing tests should see no behavior change at all. Caught and fixed a real bug before committing: the write side (BuildScalarIndexProcedure) and this read side each need to construct byte-identical TableIdentifiers to look up the same IndexIdentifier -- otherwise indexExists() here would silently and permanently return false with no error, making the whole integration a no-op that never activates. Both sides now derive it the same way, from TableIdentifier.parse(table.name()) on the core Table object, not from Spark's own catalog Identifier type. Same caveat as the previous commit: unverified against a real Spark runtime, since gradle cannot start in this sandbox at all (blocked at the local socket/daemon level, not just the domain-allowlist issue from earlier in this session). Every Iceberg-core API call (UnboundPredicate, NamedReference, Schema.findField, MetadataColumns, TableIdentifier.parse) was verified against the actual source in this repo; needs a real compile and test run before being trusted. Also adds TestScalarIndexScanPruning: functional-correctness tests (equality query returns the right row / no rows for a missing key / still correct with no index at all) plus one best-effort test using Dataset#inputFiles() to check the indexed query reads no more files than an equivalent un-indexed query -- flagged in that test's own javadoc as unverified whether inputFiles() actually reflects Iceberg's final pruned file set for this Scan implementation.
The previous version hand-rolled the whole build pipeline (compute buckets, sort, write Parquet, commit) directly in the script, in parallel to the actual production code -- meaning it could silently drift from what BuildScalarIndexProcedure actually does. It also never computed `position`, so it only ever achieved file-level pruning, not the exact-row (file, position) pruning that's the actual differentiator of a SCALAR index over the sibling Bloom filter index. This version calls the real procedure via SQL instead, so the demo exercises the real code path. It also picks a real medallion value out of the table for the lookup rather than a hardcoded one that might not exist in whatever dataset is loaded, and uses Dataset#inputFiles() to show the file count before/after rather than hand-computing the hash bucket and calling TrackingFileReader directly. Same caveat as the last two commits: not run against a real Spark session.
Table.uuid() returns java.util.UUID, not String -- ScalarIndexCommitter. commit() needs the String form. First real compile error surfaced by actually running the build, exactly as expected given this code was entirely unverified before now.
transformValueColumn() referenced the key column by its original name, but by the time it runs the earlier .select() has already renamed it to "__key" -- the original name no longer exists in that DataFrame's schema. Caught by actually running the test: UNRESOLVED_COLUMN on the UDF's input reference.
Encoders.javaSerialization() requires the target class to be public, not just Serializable -- caught by actually running the test: SparkUnsupportedOperationException, "is not a public class." BuildResult doesn't need the same fix since it's a plain driver-side return value, never passed through any Spark Dataset/Encoder API.
Spark's ClosureCleaner requires every object captured inside a UDF closure to be serializable, even for local (non-distributed) execution -- caught by actually running the read-path test: java.io.NotSerializableException: org.apache.iceberg.index.HashTransform, surfaced through BuildScalarIndexProcedure's UDF-based transform_value computation. HashTransform only wraps a single int field, so this has no real downside -- the same pattern Iceberg's own core classes (Schema, Types.NestedField) already follow for the same reason.
row_number() returns IntegerType, not LongType -- __position ended up
as an Integer-backed column, but writeLeafFilePartition read it back
via row.getAs("__position") with an inferred Long type, throwing
ClassCastException at runtime (not a compile error, since getAs()'s
type parameter is just a cast, not checked against the row's actual
schema). Fixed at the source by casting the column expression to
LongType explicitly, rather than only at the read site, so the
DataFrame's schema genuinely matches what downstream code expects.
…ning
tryPruneUsingScalarIndex() injected Expressions.equal("_file", ...) into
filterExpressions to constrain the scan to a resolved file. That list
also feeds SparkSchemaUtil.prune() (via pruneColumns) and the eventual
core Scan#filter() call, both of which bind expressions against the
real table schema via Binder -- but "_file" is a Spark-only metadata
column, not a real schema field, so binding it always threw
ValidationException: Cannot find field _file. This broke every query
that reached column pruning, not just ones an index could resolve.
Core Iceberg Scan/Expression/Binder has no concept of filtering by
metadata columns like _file at all -- enforcing this would need a Scan
decorator around planFiles()/planTasks() at the task level, a larger
follow-up. For now tryPruneUsingScalarIndex logs the resolved file but
does not enforce it; the underlying predicate is still pushed down and
applied normally, so functional correctness is unaffected.
The test built its own IndexIdentifier from the Spark catalog Identifier's namespace()/name(), but BuildScalarIndexProcedure commits under an IndexIdentifier derived from TableIdentifier.parse(table.name()) -- the core Iceberg Table's own name, chosen deliberately because SparkScanBuilder on the read side only has the core Table, not the original Spark Identifier, so both sides need a common, independently computable source. The two derivations do not necessarily produce the same TableIdentifier (table.name() can be catalog-qualified), so the test's indexExists() lookup used a different key than what was committed and always returned false. Fixed by deriving the test's IndexIdentifier the same way production code does.
Still failing after aligning both sides to derive IndexIdentifier from table.name() -- root cause was one level deeper: validationCatalog is a separate Catalog handle (e.g. a freshly-constructed HadoopCatalog, or a HiveCatalog/RestCatalog initialized under its own catalog-name property), independent of the Spark-registered catalog that BuildScalarIndexProcedure and SparkScanBuilder actually load the table through. Several Iceberg catalog implementations embed the loading catalog's own name into Table.name(), so the same physical table loaded via validationCatalog vs. via Spark produced two different name() strings, hence two non-equal TableIdentifiers, hence indexExists() always returning false in the test -- consistent with every catalog config in the parameterized suite failing identically. Fixed by loading the table via Spark3Util.loadIcebergTable(spark, tableName), the same pattern used throughout the rest of this test suite, so the test's table.name() matches what production code actually sees.
tryPruneUsingScalarIndex() previously only logged the resolved file path -- core Iceberg Scan/Expression/Binder has no concept of filtering by metadata columns, so there was no way to push the constraint through the normal Expression pipeline. Enforcing it requires intercepting task planning directly. Added FileScanTaskFilteringScan, a thin BatchScan decorator that overrides only planFiles() to filter to a resolved set of file paths, delegating everything else. This is enough because SparkPartitioningAwareScan (confirmed by reading it) only ever calls planFiles(), never planTasks(), when materializing Spark input partitions -- so no changes were needed to the scan-wrapper classes that consume it. Wired in only for the plain SELECT batch-scan path in SparkScanBuilder#buildBatchScan(); incremental-append, changelog, merge-on-read, and copy-on-write scans are left alone on purpose, since row-level operations have different correctness considerations worth their own review. Generalized tryPruneUsingScalarIndex() from acting only when exactly one file resolves, to the full set of matching files, so the optimization is not limited to single-row lookups. The zero-match case (index confirms the key is absent) is deliberately not pruned to zero files: that would be correctness-sensitive (a bug would silently return wrong empty results) rather than just missing an optimization, and is left as a documented follow-up. FileScanTaskFilteringScan self-verifies before trusting the resolved paths: it matches them against real candidate files from normal planning, and falls back to the unfiltered file set if none match -- guarding against a possible path-format mismatch between how the index recorded paths (Spark input_file_name() at build time) and how Iceberg reports them at scan time (DataFile#path()), since that agreement has not been verified across every FileIO implementation. This keeps the design invariant intact: the index must never be required for correctness, only used opportunistically for pruning. Strengthened TestScalarIndexScanPruning to assert an exact file count now that pruning is actually enforced, and added testPrunesBeyondNativeMinMaxStats, which constructs files whose min/max ranges all cover the query literal so Iceberg own manifest stats pruning cannot exclude any of them, to demonstrate the index adds real pruning value beyond native stats rather than a test that would pass with the pruning logic disabled.
…n test The full spark-extensions module run (2722 tests) came back with only 8 failures, all in TestScalarIndexScanPruning, all "expected: 1 but was: 0" on Dataset#inputFiles().length -- no regressions anywhere else, confirming the decorator itself is not the problem, the verification mechanism is. Dataset#inputFiles() returns empty for Iceberg's DataSourceV2 scans in this Spark version regardless of pruning -- it is not wired up for non-FileScan V2 sources, so both the indexed and unindexed queries always reported 0 files, which is also why the original, pre-existing "<=" comparison always passed trivially before this change strengthened it to an exact count. Separately, the resultDataFiles SQL metric (ScanMetricsUtil, called from ManifestGroup) is recorded inside the wrapped scan's own planFiles() -- upstream of this decorator's filter -- so it reflects native manifest-stats pruning only, never this decorator's additional restriction. Neither is a usable signal for this class specifically. Removed both SQL-level file-count assertions and the now-redundant testInputFilesReducedAfterIndexBuild test (its distinguishing signal was unreliable, and its correctness check duplicated an existing test). Added TestFileScanTaskFilteringScan, a focused unit test against a real (if minimal) Iceberg table built via the existing TestTables test helper, verifying planFiles() directly: filters to a single allowed path, filters to multiple allowed paths, falls back to the unfiltered set when no allowed path matches any real candidate (the self-verifying safety net), and delegates other methods unchanged. This is a more direct and deterministic proof of the decorator's own logic than any Spark Dataset-level signal available here. Also switched FileScanTaskFilteringScan's path comparison from the deprecated ContentFile#path() to #location().
Closeable#close() declares throws IOException, so try-with-resources on CloseableIterable requires the enclosing method to declare or catch it -- missed across all three try-with-resources blocks.
github-actions
Bot
added
API
spark
core
data
Specification
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Reference implementation of the SCALAR index type defined in #16961.
Supersedes #17426, which the stale bot auto-closed for 30 days of inactivity (explicitly not a
judgement on the merits) before this branch's Spark integration existed. Continuing that work here
with a full Spark integration added since.
What is included
Spec (format/index.md from #16961) + Core/API/Data/Spark layers:
Layer 0 — Data model: IndexMetadata, IndexSnapshot interfaces and JSON parsers
Layer 1 — Storage: IndexCatalog, IndexMetadataIO, InMemoryIndexCatalog
Layer 2 — Tracking file: TrackingFileWriter + TrackingFileReader (Avro, same as Iceberg manifests)
Layer 3 — Build: HashTransform, LeafFileMetadata, ScalarIndexCommitter, LeafFileEntry/Writer/Reader
Layer 4 — Spark integration:
CALL system.build_scalar_index(table, columns, transform, options)populates an index from a live tableself-verifying: falls back to the unfiltered file set if resolved paths don't match real
candidates, so a bug here can only miss an optimization, never return a wrong result
Test status
Core/API/Data: unit tests from earlier layers (index metadata round-trip, tracking/leaf file
read-write, in-memory catalog, end-to-end build+commit+lookup wiring).
Spark: TestBuildScalarIndexProcedure, TestScalarIndexScanPruning, and TestFileScanTaskFilteringScan
green; full spark-extensions module regression run (2722 tests) confirms no unrelated breakage.
Known gaps / explicit non-goals for this pass
deferred since a bug there would silently return wrong (empty) results, not just miss an
optimization
scans are untouched
not yet durable across restarts
build_scalar_indexonly supports full rebuilds -- no incremental/append-only update path yet.Huaxin Gao's Primary Key Index for Apache Iceberg
proposal (see "Relationship to the PK Index proposal" below) found full-leaf rebuild maintenance
falls behind realistic checkpoint budgets under CDC-style churn at large key counts for a closely
related index, and lays out concrete incremental-refresh options (append-only with periodic
cleanup, synchronous update, or hybrid) that this index could adopt directly
Relationship to the PK Index proposal
Thanks @huaxingao for putting together the PK design, it already has covered quite a few categories, specially delete.
I’ll further review it and find delta between the two design docs, and we can converge iI, appreciate your time!
Huaxin Gao's Primary Key Index for Apache Iceberg
proposes another concrete index type under the same Secondary Index framework (#16961), looks like both has map key -> (file, position) and both use scan-time file pruning as a core mechanism.
Builds on