[ESQL] Introduce pluggable external datasource framework

[costin]

Add a modular framework for querying external data sources from ESQL via the EXTERNAL command. The goal is to enable ESQL to query data residing outside Elasticsearch indices, including datalakes (Iceberg), cloud storage (S3), HTTP endpoints, and local files, using various formats.

The architecture consists of two pluggable layers:

• Storage layer: protocol-agnostic access to storage systems

  • StorageProvider/StorageObject SPIs with HTTP, S3, and local filesystem implementations
  • Range-based reads for efficient columnar format access

• Format layer: data parsing into ESQL Page batches

  • FormatReader SPI with CSV and Parquet implementations
  • Schema inference and column projection support

Additional components:

• Iceberg datalake adapter with catalog integration and filter pushdown
• ExternalSourceOperatorFactory combining storage and format layers
• Parser and analyzer extensions for EXTERNAL relation syntax
• Physical plan mapping and LocalExecutionPlanner integration
• Comprehensive test suite including S3 fixture-based integration tests

Developed using AI-assisted tooling

[elasticsearchmachine]

Pinging @elastic/es-analytical-engine (Team:Analytics)

[idegtiarenko]

I wonder if we should resolve external sources as part (or even inside) of indexResolver?
Current approach indirectly implies external sources are separate paths, presumably only in FROM. If we can make them part of IndexPattern (or just pattern) we could eventually make joins and subqueries with external sources as well as select from mix of indices and external sources at once.

[costin]

I worked on minimal changes to this area as my goal for the PR is end-to-end testing.
That said, as the discovery paths are separate, I would keep the indexResolver, which is fairly tied to FieldCaps & co, in place and work on generifying the PreAnalyzer/Analysis phase.

[idegtiarenko]

I wonder if we should rely on org.elasticsearch.common.blobstore.BlobStore instead?
It is used by snapshot and already implemented for s3, gcs, azure, fs and probably others.

This will allow us to leverage existing infrastructure for accessing external sources rather than creating new infrastructure.
It is part of the server package so it should not create any extra unwanted dependencies (unless eventually we want to abstract that as well).

[costin]

I started with Blob however it adds both read and write methods so ended up with the above interface which is read-onlly focused. Definitely something to look into closer to see whether the two interfaces could be unified (as well as the code behind it).

[idegtiarenko]

Do we plan to use it only to detect a format for the object?
I think we need to keep in mind structures that spans multiple files with multiple extensions (such as lucene indices as well).

Also

Suggested change

	List<String> fileExtensions();
	Collection<String> fileExtensions();

for more flexibility

[costin]

In this PR I started simple, with one format per provider. This can be extended to have mixed formats through hierarchical provider that.

[idegtiarenko]

I think it is worth having ExternalSource object. For now it could only have a name but later it could be used to describe what kind of optimizations (such as pushdown) it supports.

[costin]

👍

[idegtiarenko]

Suggested change

	String tablePath();
	String dataPath();

Also do we expect it to be spread across multiple files (again thinking about lucene index like data structures with multiple files with different functions backing the actual data)

[costin]

I believe you're touching on two topics here:

1. logical storage - which is composed of multiple files that work together as a unit
2. heterogeneous data - different file formats and providers that can be seen as a unified table

Neither is covered in this PR which focused on the SPI primitive where a logical storage = physical one. Once settled, we should be able to assemble the logical storage with this primitive.
Point 2 is conceptually simliar to our IndexResolver as you already pointed out however there's a lot more work on the planner side to efficiently do data loading and processing based on the overall plan and cost analysis, which is way outside the current iteration scope.

[idegtiarenko]

We use dedicated thread pools for snapshot read operations:

elasticsearch/server/src/main/java/org/elasticsearch/threadpool/DefaultBuiltInExecutorBuilders.java

Lines 147 to 160 in d8c3c6f

	result.put(
	ThreadPool.Names.SNAPSHOT,
	new ScalingExecutorBuilder(ThreadPool.Names.SNAPSHOT, 1, maxSnapshotCores, TimeValue.timeValueMinutes(5), false)
	);
	result.put(
	ThreadPool.Names.SNAPSHOT_META,
	new ScalingExecutorBuilder(
	ThreadPool.Names.SNAPSHOT_META,
	1,
	Math.min(allocatedProcessors * 3, 50),
	TimeValue.timeValueSeconds(30L),
	false
	)
	);

SNAPSHOT for actual lucene files and SNAPSHOT_META for fairly small files describing the snapshot structure.

I believe we should revisit overall esql thread pool usage and split operations into compute/memory heavy (actual transformations and aggregations) and read heavy. With this change I believe we should split read heavy further into reading local lucene files and remote data source files.