Adds a bloom filter index for the specified column.

Bloom filters are probabilistic data structures that provide:

• Guaranteed NO false negatives (if filter says "no", value definitely absent)
• Configurable false positive rate (if filter says "yes", value MIGHT be present)
• Space-efficient storage (approximately 10 bits per element at 1% FPR)

Binary columns cannot be bloom-indexed. Values are canonicalized to a string before hashing, and Spark materializes BinaryType as a JVM Array[Byte] whose toString is identity-based ([B@1b6d3586), so the same bytes hash to a different token on every occurrence. Such a filter would match nothing, ever. Columns that contain a binary field at any depth are rejected for the same reason. Use addIndex on the column instead — regular indexes compare values, not their string form, and work correctly on binary data.

Adds a computed index derived from a SQL expression.

The expression is evaluated during update to produce a virtual column whose distinct values are stored in the index. Idempotent: calling again with the same name is a no-op.

Adds an exploded field index for a nested field inside an array column.

During update, each element of arrayColumn is exploded, the fieldPath is extracted, and the distinct values are stored under asColumn in the index. Joins on asColumn will locate files containing any matching array element. Idempotent: calling again with the same asColumn is a no-op.

Adds files to the index's file list for future indexing. Acquires a file list lock to prevent concurrent modifications.

Adds a stable filename column to the DataFrame.

Uses input_file_name() to capture the source file path. For single-file reads where Spark may report an empty filename, falls back to the known file path.

Adds a regular (array-of-distinct-values) index for the specified column.

Idempotent: calling again with the same column is a no-op.

Adds a range index for the specified column.

Range indexes store min/max values per file, enabling file pruning at query time. Files whose [min, max] range does not overlap with the queried values are skipped.

Adds a temporal index for the specified column using a timestamp for versioning.

When joining on a temporal index column, only the latest version (by timestamp) of each value is returned. This is useful when multiple files contain the same entity at different points in time.

The timestamp may be a nested path such as meta.updatedAt, since it is only read during index construction and deduplication. The value column must be top-level: it is persisted as a column of the index table under its own name, so a dotted path could not be read back.

Performs pre-flight analysis on unindexed files to determine optimal batching strategy.

Reads the source files, applies computed indexes, and counts distinct values per indexed column per file. The resulting FileAnalysis objects are used by createOptimalBatches to group files into batches that stay under the largeIndexLimit.

If no storage columns are configured (e.g., only bloom or range indexes), returns trivial analyses with zero distinct counts.

Each column is counted on the same basis its builder uses, because these counts also drive classifyLargeFiles. Regular, computed, and temporal columns are counted before exploded fields are applied: applyExplodedFields uses an inner explode, which drops rows whose array is null or empty and would undercount every other column. Temporal columns add one for a present null value, matching the collect_set of struct(value, max_ts) in buildTemporalIndexes, which retains a struct for the null group.

Appends large index data to per-column Delta tables under large_indexes/.

For each column with at least one large file, the values of those files are read straight from the source DataFrame as distinct (filename, value) rows and written to large_indexes/{column}/. Values are never collected into a per-file array first, so a file's cardinality is bounded by what Delta can store rather than by what fits in an executor-side array. Before appending, any existing rows for the same filenames are removed via Delta MERGE to prevent duplicates (important during column backfill or re-indexing).

Appends the processed DataFrame to the staging Delta table.

Columns belonging to a large (file, column) pair are already null at this point: the index builders skip them and their values were streamed to large_indexes/ instead. Staging deliberately does not re-derive largeness from array size. Doing so would be a second, independent decision, and any disagreement with classifyLargeFiles would null out an array whose values were never written anywhere else, silently losing them.

Applies column selection if columns have been specified via select(). Only reads the explicitly selected columns.

Applies computed indexes to a DataFrame by adding computed columns.

Each computed index is a Spark SQL expression stored in metadata. The expression is evaluated via expr() and added as a new column. If no computed indexes are configured, the DataFrame is returned unchanged.

Applies exploded field transformations to a DataFrame.

For each configured exploded field, extracts the nested field path from the array column and explodes it into a top-level column. This is used on the read/join path to prepare data for joining on exploded field values.

Note that explode() (not explode_outer()) is used intentionally — rows with null or empty arrays are dropped because they contain no matchable values for the join. This is consistent with the index build path which also excludes null/empty arrays.

Column name prefix for auto-bloom filter storage in the main index table.

False positive rate for auto-bloom filters on large index columns. When a column exceeds largeIndexLimit, an auto-bloom filter is built with this FPR to enable file pruning at query time. Reads from spark.ariadne.autoBloomFpr configuration (default: 0.01).

Returns the set of auto-bloom storage column names (each prefixed with auto_bloom_).

Returns the expression selecting the bloom-hashable scalar out of a columnValueRows row.

Every column type but temporal yields the value directly. Temporal rows carry a (value, max_ts) struct, so the filter is folded over the value field alone to match the bare scalars that queries probe with.

Optional threshold for auto-compaction. When set, every Delta table backing the index (main index and each large index) is compacted during an update once the persisted batches-since-compact counter meets or exceeds this value. The counter tracks update batches, not Delta log files. Reads from spark.ariadne.autoCompactThreshold configuration (default: not set).

Counter tracking batches processed since the last auto-compaction.

Incremented after each batch in updateBatched and reset to zero after each compaction cycle or at the start of each update call to prevent stale counts from a previous update from triggering premature compaction.

Column name prefix used when storing bloom filter binary data in the index Delta table.

Returns the set of source column names that have bloom indexes configured.

Returns the bloom index configurations from the current index metadata.

Returns the set of storage column names used for bloom filter binary data.

Each name is the source column name prefixed with bloomColumnPrefix.

Builds auto-bloom filters for columns that have at least one large file.

A column becomes auto-bloom the first time any file contributes at least largeIndexLimit distinct values to it. From then on every file gets a filter, stored in the main index under the auto_bloom_ prefix, so queries can cheaply skip files before touching large_indexes/.

Filters are folded directly from the source (filename, value) rows with a streaming aggregator rather than from a collected array. Large files never have an array to read from, and building from the rows means a column's cardinality is limited only by what the filter itself costs (~1.2 bytes per distinct value at 1% FPR).

Builds bloom filter indexes for all configured bloom columns.

For each BloomIndexConfig, this method:

1. Reduces the input to distinct (filename, value) pairs
2. Computes each file's distinct value count so its filter can be sized individually
3. Folds the values into a bloom filter with BloomFilterAggregator
4. Joins the resulting bloom column back onto the accumulating DataFrame

The aggregation is streaming: values are hashed into the filter and discarded, so executor memory scales with the size of the filter (~1.2 bytes per distinct value at 1% FPR) rather than with the number of values held as boxed JVM objects. A bloom-indexed column is therefore not bounded by what fits in an executor-side array.

If no bloom indexes are configured, returns a distinct filename-only DataFrame.

Builds exploded field indexes and joins them onto the result DataFrame.

For each configured exploded field index, extracts the nested field path from the array column, explodes it, collects distinct values back into an array per file, and joins the result onto resultDf.

Builds range indexes storing Struct(min, max) per file.

For each range index configuration, groups by filename and computes the min and max of the column per file. The result is a struct column named range_{column}.

Builds regular (array-aggregated) indexes for all configured regular and computed columns.

Groups the input data by filename and collects distinct values into arrays via collect_set. If no regular or computed indexes are configured, returns a distinct filename-only DataFrame.

Folds a column of values into one serialized bloom filter per file.

The aggregation is streaming: values are hashed into the filter and discarded, so executor memory scales with the size of the filter (~1.2 bytes per distinct value at 1% FPR) rather than with the number of values held as boxed JVM objects. A bloom-backed column is therefore not bounded by what fits in an executor-side array.

Values are reduced to distinct (filename, value) pairs first, so each value is inserted exactly once.

Builds temporal indexes storing Array[Struct(value, max_ts)] per file.

For each temporal index configuration, groups by (filename, value_column) to find the maximum timestamp per value per file, then collects the (value, max_ts) structs into an array per file. This enables temporal deduplication at query time.

Converts a column of any type to the canonical string form used for bloom hashing.

Build and query sides must agree exactly on this representation: a bloom filter guarantees no false negatives, but only if the same value hashes to the same string in both places. The query side stringifies driver-side JVM objects with toString, so this wraps the value in a single-element array and applies toString inside a UDF. Spark converts array elements to the same external JVM types (java.sql.Timestamp, java.math.BigDecimal, and so on) that the driver sees, making the two paths byte-identical for scalar types.

This does not hold for types whose toString is not value-based. Array[Byte].toString is an identity hash, so two equal binary values stringify differently on the build and query sides and every probe misses. Binary columns are therefore rejected by Index.addBloomIndex and excluded from auto-bloom. Nested array and map types are untested and carry the same risk.

A plain cast(StringType) would not be safe here — Spark renders a timestamp as 2024-01-01 00:00:00 while java.sql.Timestamp.toString renders 2024-01-01 00:00:00.0, which would silently produce false negatives.

Classifies each (file, column) pair as large or inline using pre-flight distinct counts.

Produces the (filename, value) rows for an index column, one row per occurrence.

Rows are produced directly from the source data without aggregating into arrays. This is the value source for both large_indexes/{column} and auto-bloom filters. Callers are responsible for any distinct or filename filtering they need.

Regular and computed indexes project the column directly; exploded-field indexes explode the configured nested path; temporal indexes reduce to one (value, max_ts) struct per distinct value.

Compacts all Delta tables belonging to this index using OPTIMIZE. Acquires the update lock to prevent concurrent modifications.

Compacts all Delta tables (main index and large index tables) using Delta OPTIMIZE.

Runs Delta Lake's OPTIMIZE command to consolidate small files into larger ones, improving read performance. Processes the main index table first, then each large index column table.

Consolidates staged data into the main index table.

Delegates to consolidateMainStaging which performs a Delta MERGE (upsert) of all staged rows into the main index, then deletes the staging table. Logs the total time taken.

Creates a base DataFrame from the provided files using the stored format and schema. Applies any read options configured in the index metadata.

Returns an empty DataFrame (with the stored schema) if all file paths are blank after normalization.

Groups files into batches whose aggregate maxDistinctCount stays at or below largeIndexLimit.

The algorithm sorts files by maxDistinctCount (largest first) and packs them sequentially into batches. When adding a file would push the batch total past the limit, a new batch is started. Files that individually exceed the limit are placed into single-file batches to guarantee progress.

When true, logs detailed diagnostics during join operations including partition counts, physical plans, and per-phase timing. Reads from spark.ariadne.debug configuration (default: false).

Deletes a path recursively from the filesystem.

Deletes the specified files from the index, large index tables, and file list.

Acquires the update lock, removes matching rows from the main index Delta table, all large index Delta tables, and the FileList. If a filename doesn't exist in the index, it is silently ignored.

Returns a io.delta.tables.DeltaTable if the path exists and contains valid Delta metadata.

Three outcomes are possible:

• the path holds a valid Delta table — Some(DeltaTable);
• the path is absent, or exists but is an empty directory — None, so the caller may create the table there;
• the path exists, is non-empty, and holds no readable Delta metadata — dev.cjfravel.ariadne.exceptions.InvalidDeltaTableException.

The third case is reported rather than repaired. Ariadne cannot distinguish an abandoned partial write of its own from data written by something else, so it never deletes the directory; cleanup is left to the caller, who can inspect the contents first. An empty directory is treated as absent because there is nothing to inspect and nothing to lose.

Runs migration preflight while the caller holds the update lock.

Checks if a path exists on the filesystem.

Returns the file format of the indexed data (e.g., "parquet", "csv", "json").

Hadoop FileSystem instance resolved from the storagePath URI and the Spark Hadoop configuration. Used for all filesystem operations (existence checks, reads, deletes, lock files).

Gets candidate files from the auto-bloom filter using values extracted from a DataFrame.

Delegates the bounded query-side collect to BloomFilterOperations.collectProbeValues, the same helper the explicit bloom path uses, then probes via getAutoBloomCandidates. The two bloom kinds differ only in how a null filter is interpreted: a missing auto-bloom means the file was never large enough to get one and must stay a candidate, whereas a missing explicit bloom means the file held no values for the column and cannot match.

Computes file sizes in bytes for the given files using the Hadoop FileSystem.

Files that cannot be found or read are silently skipped with a warning log.

Writes the values of every large (file, column) pair into its per-column Delta table.

Checks if a file is tracked by this index's file list.

Helper function to load the index.

On first call, checks for and performs any needed exploded field column migration (pre-0.1.1 indexes stored columns under array_column names).

Hadoop path for the main index Delta table ({storagePath}/index/).

Optional number of partitions to use when repartitioning the index DataFrame during joins. When set, the index DataFrame is repartitioned before expensive operations like explode to reduce per-executor memory pressure and avoid FetchFailedExceptions on large indexes. Reads from spark.ariadne.indexRepartitionCount configuration (default: not set).

Returns all column names that can be used in joins across all index types.

Includes regular, computed, exploded field, bloom, temporal, and range index columns.

Joins the indexed data with the provided DataFrame.

Locates relevant data files via the index, reads them, applies temporal deduplication if configured, and joins the result with the provided DataFrame.

What this returns: the minimal complete dataset for the join. Every indexed value that matches df is found, and index-side rows absent from df are discarded. Because pruning selects whole files, unmatched index-side rows that share a file with a match are still read, so the result can include incidental rows and can change after Index.compact. Join types whose result is defined by those rows — left, full and left_anti, since the index is on the left here — are rejected rather than returning a file-layout-dependent answer.

Locates and reads indexed data files relevant to the given DataFrame.

Uses the index to identify which files contain matching values, then reads those files into a lazy DataFrame. If no matching files are found, returns an empty DataFrame produced through the same read path as the populated branch, so computed indexes, exploded fields and any active select() are applied and the schema matches either way. The actual row-level filtering happens in the subsequent join in join.

Logs data pruning metrics (file count, data size saved) when available, and includes detailed file-level debug information when debug mode is enabled.

Returns the set of column names that have large index Delta tables.

Maximum number of distinct values per file before an index column is promoted to a "large index" (stored in a separate exploded Delta table). Reads from spark.ariadne.largeIndexLimit (default: 500000).

If the configured value is not a valid positive number, a warning is logged and the default of 500,000 is used.

Hadoop root path for large index Delta tables ({storagePath}/large_indexes/).

Returns a unified (filename, colName) DataFrame for a regular index column by exploding the main index arrays and unioning with any large index rows that exist for that column.

Returns a unified (filename, _value, _max_ts) DataFrame for a temporal index column by exploding the main index struct arrays and unioning with any large index rows that exist for that column.

Locates files matching the given index values across all index types.

Queries regular, bloom, temporal, range, and auto-bloom indexes in parallel and intersects results with AND semantics. Each index type independently returns candidate files, and only files present in all queried categories are included in the final result.

Locates files based on a DataFrame containing join column values. Handles both regular indexes and bloom filter indexes.

Locates files that might contain any of the given values using bloom filters.

The probe runs as a distributed Spark filter via probeBloomFilters: candidate values are broadcast to the executors and only matching filenames are returned to the driver. Bloom filter binary data is never collected.

Locates files that might contain values from a DataFrame using bloom filters.

Collects distinct non-null values from the specified column of valuesDf, then delegates to locateFilesWithBloom for the actual bloom filter check.

Maximum total time in seconds to wait for lock acquisition before failing. Reads from spark.ariadne.lockMaxWait configuration (default: 3600).

Refresh the update lock every N batches during updateBatched. Reads from spark.ariadne.lockRefreshInterval configuration (default: 1).

Base interval in seconds between lock acquisition retries (exponential backoff applied). Reads from spark.ariadne.lockRetryInterval configuration (default: 60).

Seconds since lastRefreshedAt before a lock is considered stale and eligible for auto-heal (forcible acquisition by another process). Reads from spark.ariadne.lockTimeout (default: 1800).

Log4j logger shared by all Ariadne components. Uses the "ariadne" logger name. Marked @transient lazy so that Index (a case class) remains serializable across Spark stages — Log4j loggers are not Serializable.

Maps join column names to their corresponding storage column names.

Bloom filter columns are prefixed with the bloom prefix, range columns are prefixed with "range_", exploded field columns are mapped to their backing array column, and all other columns map to themselves.

Triggers compaction if the auto-compact threshold has been reached.

Checks the batchesSinceCompact counter against the configured autoCompactThreshold. If the threshold is met, compacts all Delta tables and resets the counter to zero. The counter is persisted to metadata so it survives across Spark jobs.

Conditionally repartitions a DataFrame if indexRepartitionCount is configured. This helps avoid FetchFailedExceptions when working with very large index DataFrames by spreading data across more partitions before expensive operations like explode.

Checks if metadata exists in the storage location.

Hadoop path for the metadata file.

Builds auto-bloom filters for columns that already have a large_indexes/ table but no filter in the main index.

Filters are folded from autoBloomValueRows, covering files whose values are inline as well as those in the large index. Files that hold at least one non-null value for the column end up with a filter. Files that hold none keep a null filter, which the query side treats as an unconditional candidate (includeNullFilters = true), so every file is still represented and a query can treat an empty candidate set as a definitive no-match.

Metadata is written only after every column has been backfilled, so a failure part way through leaves the storage version unchanged and the next preflight repeats the work.

Migrates pre-0.1.1 indexes that stored exploded field columns under array_column names to the current as_column naming convention.

Detection: reads the main and staging Delta table schemas and checks whether any ExplodedFieldMapping.array_column appears as a column while the corresponding as_column does not. When this pattern is found, each table is rewritten with renamed columns, and any large-index directories named by array_column are similarly migrated.

This method is idempotent and is invoked only by storage migration preflight while the update lock is held.

Rewrites large index tables whose value column still uses a legacy name so they store it under the canonical name.

loadLargeIndex resolves the legacy shape on read, but appendToLargeIndex writes under the canonical name with mergeSchema, which would widen a legacy table into two value columns. Rewriting once removes that divergence.

Opens an input stream to read from a path on the filesystem.

Probes every file's bloom filter for bloomColumn against values, entirely on the executors.

This is the memory-safe counterpart to collecting bloom filter binary data to the driver. The candidate values are broadcast, the mightContain test runs as a distributed filter, and only the surviving filenames are returned to the driver. Driver memory is therefore bounded by the number of matching files rather than by the total serialized size of the bloom filters.

This distinction matters most for auto-bloom columns. Once any one file crosses largeIndexLimit distinct values (500,000 by default), every file in that column gets a filter, each sized from that file's own distinct count; the largest can reach roughly 585 KiB at the default 1% FPR. Collecting one per file would scale driver memory with the number of indexed files.

Returns the set of range index storage column names (each prefixed with range_).

Forces a reload of metadata from the metadata.json file on disk.

Replaces the in-memory cached metadata with a freshly deserialized copy. This is called automatically by metadata on first access; callers may invoke it explicitly to pick up external changes.

When true, applies indexRepartitionCount repartitioning to the data files read during joinDf. When false, data files keep their natural parquet partitioning. Disable when column selection significantly reduces data volume, making the repartition shuffle more expensive than useful. Reads from spark.ariadne.repartitionDataFiles configuration (default: false).

Rejects a column whose values cannot be canonicalized into a stable bloom filter token.

Bloom filters hash the toString form of each value. Spark materializes BinaryType as a JVM Array[Byte], whose toString is identity-based ([B@1b6d3586), so the same bytes produce a different token on every occurrence — even for two references to the same object read at different times. A filter built over such a column is internally consistent but matches nothing, so every query prunes away every file. Nested binary is rejected for the same reason.

The check is skipped when no schema is stored, matching the surrounding column validations: the type cannot be established without one.

Enforces that a column carries at most one index type.

Checking every type in one place keeps the matrix symmetric by construction: a column carrying two index types can produce wrong results at query time rather than an error, and per-method lists of checks make detection depend on registration order wherever an entry is missing. A new index type therefore cannot reintroduce a one-directional gap.

Callers must invoke this only after confirming the column is not already registered as newType, so that re-adding the same index type remains idempotent.

Rejects nested (dotted) paths for columns that become index column names.

An indexed value column is persisted as a column of the index table under its own name, and is later read back with col(name). A dotted path such as meta.userId would be written as a literal column name but read back as nested field access, so such an index can never be built or queried.

SchemaHelper.fieldExists resolves dotted paths, so without this guard the configuration is accepted and persisted to metadata.json, and only fails later during update with an opaque Spark analysis error.

This restriction applies to the indexed value column only. Temporal timestamp columns may be nested, because they are never persisted under their own name.

Safely destroys a broadcast variable, logging but swallowing any exception.

Intended for use inside finally blocks where a cleanup failure must not mask the original exception propagating out of the try block. Tolerates a null broadcast.

Selects specific columns for optimized reading.

When set, only the selected columns (plus any join columns) are read from data files, reducing I/O. Returns this Index for method chaining.

Implicit SparkSession that must be provided by the mixing class.

Number of batches to process before consolidating staged data into the main index. This provides fault tolerance for large index builds - if a job fails, work is preserved up to the last consolidation point. Reads from spark.ariadne.stagingConsolidationThreshold configuration (default: 50).

Hadoop path for the staging Delta table ({storagePath}/staging/).

The staging table accumulates batch results during update and is merged into the main index by consolidateStaging.

Returns a DataFrame of per-column index statistics and total file count.

For each indexed column, computes statistics on the array length (number of distinct values per file): min, max, avg, median, and standard deviation. Also includes the total number of indexed files.

Returns an empty DataFrame if no index table exists. If storageColumns is empty (e.g., only bloom or range indexes), the result contains zero stat rows.

Returns the set of all storage column names across regular, computed, exploded-field, and temporal index types.

This is used internally to determine which columns to check for large-index separation and to build aggregation expressions.

Path to the storage location of the index.

Returns the stored schema of the index.

Parses the JSON schema string from metadata into a Spark StructType.

Updates the index with new files and backfills newly added columns.

Processes all unindexed files registered via addFile, using intelligent batching based on pre-flight analysis. Also backfills existing files when new index columns have been added since the last update.

Consistency

A column's values for a given file live in exactly one place: inline in the main index table, or — once the file contributes at least largeIndexLimit distinct values — in large_indexes/{column}/. Queries read both and union the results, so this exclusivity is what keeps a file from being counted twice or missed entirely.

That invariant is guaranteed for an update that completes successfully. The two stores are separate Delta tables written in separate commits, so an update killed part-way can leave them disagreeing about a file. In the usual case the file is then still absent from the main index, so it remains unindexed and the next update reprocesses it and restores consistency by itself. The exception is a column backfill over a file that was already indexed: its stale large-index rows are deleted before the replacements are written, and a crash in that window leaves the file present in the main index — hence not reprocessed — with its large values gone. Re-running the update does not detect this; rebuilding the index does.

This is a known and accepted limitation, recorded here rather than worked around. Callers that cannot tolerate it should treat an interrupted update as a signal to rebuild rather than resume.

Lock path for index update and storage migration operations.

Vacuums all Delta tables belonging to this index to remove old files. Acquires the update lock to prevent concurrent modifications.

A retentionHours of zero or less additionally disables Delta's retention safety check (spark.databricks.delta.retentionDurationCheck.enabled) for the duration of the call, because Delta otherwise refuses any retention below 168 hours. That check exists to stop a vacuum from deleting files a concurrent reader or writer is still using, so vacuum(0) is only safe when nothing else is touching the index — the update lock excludes other Ariadne writers, but not readers and not non-Ariadne access to the same paths.

Vacuums all Delta tables (main index and large index tables) to remove old files.

Uses Delta Lake's VACUUM command to delete data files no longer referenced by the transaction log. Temporarily disables the retention duration safety check when retentionHours is zero or negative.

Writes metadata to the metadata.json file and updates the in-memory cache.

Creates the parent directory if it does not exist. JSON is written and validated at a unique sibling path, then replaced through Hadoop's overwrite rename operation. The replacement is atomic on filesystems with atomic rename semantics (such as HDFS); object-store connectors may implement rename as copy/delete and cannot provide the same guarantee.