Implicit SparkSession that must be provided by the mixing class.

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.

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 (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.

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).

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.

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 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/).

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.

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.

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.

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

Base path on the Hadoop-compatible filesystem where all Ariadne index data is stored. Each index creates a subdirectory under this path.

Reads from spark.ariadne.storagePath (required — no default).

Returns the stored schema of the index.

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

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.