Acquires the lock for the given correlation ID.

If the lock file already exists, this method will either heal a stale or corrupt lock, or enter a retry loop until the lock becomes available or lockMaxWait is exceeded.

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

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

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.

Checks if a path exists on the filesystem.

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

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

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.

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.

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

Refreshes the lock's lastRefreshedAt timestamp to prevent stale-lock healing.

Long-running operations should call this periodically (more frequently than lockTimeout) to signal that the lock holder is still active.

Releases the lock if it is held by the given correlation ID.

If the lock file does not exist or belongs to a different correlation ID, a warning is logged and no action is taken.

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

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.

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

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