ESQL: Make datasources plugins lazy (elastic#142815)

Defer heavy plugin factory methods (storageProviders, formatReaders,
connectors, tableCatalogs) until first query access so that AWS SDK,
Parquet, and Arrow classes are not loaded at node startup.

DataSourcePlugin: added supportedSchemes/Formats/Extensions/Catalogs
and supportedConnectorSchemes for cheap capability declarations
DataSourceCapabilities: new record aggregating all plugin capabilities
at startup without triggering classloading
DataSourceModule: lazy delegating factories per-plugin via
LazyPluginState, LazyConnectorFactory, LazyTableCatalogWrapper
FormatReaderRegistry: added registerExtension for pre-registration
without lazy init; removed fallback init-all-formats path
ExternalSourceResolver: early scheme validation via capabilities;
introduced UnsupportedSchemeException replacing string matching
FileSourceFactory: canHandle uses hasExtension instead of byExtension
GrpcDataSourcePlugin: migrated to supportedConnectorSchemes
DataSourceModuleLazyLoadingTests: spy-based tests verifying per-plugin
isolation and lazy loading guarantees

Developed using AI-assisted tooling

Author: costin

docs/changelog/142815.yaml

@@ -0,0 +1,5 @@
+area: ES|QL
+issues: []
+pr: 142815
+summary: Make datasources plugins lazy
+type: enhancement

x-pack/plugin/esql-datasource-csv/src/main/java/org/elasticsearch/xpack/esql/datasource/csv/CsvDataSourcePlugin.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.xpack.esql.datasources.spi.FormatReaderFactory;
 
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Data source plugin that provides CSV format support for ESQL external data sources.
@@ -36,6 +37,16 @@
  */
 public class CsvDataSourcePlugin extends Plugin implements DataSourcePlugin {
 
+    @Override
+    public Set<String> supportedFormats() {
+        return Set.of("csv");
+    }
+
+    @Override
+    public Set<String> supportedExtensions() {
+        return Set.of(".csv", ".tsv");
+    }
+
     @Override
     public Map<String, FormatReaderFactory> formatReaders(Settings settings) {
         return Map.of("csv", (s, blockFactory) -> new CsvFormatReader(blockFactory));

x-pack/plugin/esql-datasource-gcs/src/main/java/org/elasticsearch/xpack/esql/datasource/gcs/GcsDataSourcePlugin.java

@@ -15,6 +15,7 @@
 
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 
 /**
  * Data source plugin providing Google Cloud Storage support for ESQL.
@@ -28,6 +29,11 @@
  */
 public class GcsDataSourcePlugin extends Plugin implements DataSourcePlugin {
 
+    @Override
+    public Set<String> supportedSchemes() {
+        return Set.of("gs");
+    }
+
     @Override
     public Map<String, StorageProviderFactory> storageProviders(Settings settings) {
         StorageProviderFactory gcsFactory = new StorageProviderFactory() {

x-pack/plugin/esql-datasource-grpc/src/main/java/org/elasticsearch/xpack/esql/datasource/grpc/GrpcDataSourcePlugin.java

@@ -13,13 +13,19 @@
 import org.elasticsearch.xpack.esql.datasources.spi.DataSourcePlugin;
 
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Registers the Arrow Flight connector for ESQL.
  * Handles {@code flight://} and {@code grpc://} URIs for columnar data streaming via gRPC.
  */
 public class GrpcDataSourcePlugin extends Plugin implements DataSourcePlugin {
 
+    @Override
+    public Set<String> supportedConnectorSchemes() {
+        return Set.of("flight", "grpc");
+    }
+
     @Override
     public Map<String, ConnectorFactory> connectors(Settings settings) {
         return Map.of("flight", new FlightConnectorFactory());

x-pack/plugin/esql-datasource-http/src/main/java/org/elasticsearch/xpack/esql/datasource/http/HttpDataSourcePlugin.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.xpack.esql.datasources.spi.StorageProviderFactory;
 
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ExecutorService;
 
 /**
@@ -35,6 +36,11 @@
  */
 public class HttpDataSourcePlugin extends Plugin implements DataSourcePlugin {
 
+    @Override
+    public Set<String> supportedSchemes() {
+        return Set.of("http", "https", "file");
+    }
+
     @Override
     public Map<String, StorageProviderFactory> storageProviders(Settings settings, ExecutorService executor) {
         return Map.of(

x-pack/plugin/esql-datasource-iceberg/src/main/java/org/elasticsearch/xpack/esql/datasource/iceberg/IcebergDataSourcePlugin.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.xpack.esql.datasources.spi.TableCatalogFactory;
 
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Data source plugin that provides Iceberg table catalog support for ESQL external data sources.
@@ -37,6 +38,11 @@
  */
 public class IcebergDataSourcePlugin extends Plugin implements DataSourcePlugin {
 
+    @Override
+    public Set<String> supportedCatalogs() {
+        return Set.of("iceberg");
+    }
+
     @Override
     public Map<String, TableCatalogFactory> tableCatalogs(Settings settings) {
         return Map.of("iceberg", s -> new IcebergTableCatalog());

x-pack/plugin/esql-datasource-parquet/src/main/java/org/elasticsearch/xpack/esql/datasource/parquet/ParquetDataSourcePlugin.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.xpack.esql.datasources.spi.FormatReaderFactory;
 
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Data source plugin that provides Parquet format support for ESQL external data sources.
@@ -36,6 +37,16 @@
  */
 public class ParquetDataSourcePlugin extends Plugin implements DataSourcePlugin {
 
+    @Override
+    public Set<String> supportedFormats() {
+        return Set.of("parquet");
+    }
+
+    @Override
+    public Set<String> supportedExtensions() {
+        return Set.of(".parquet");
+    }
+
     @Override
     public Map<String, FormatReaderFactory> formatReaders(Settings settings) {
         return Map.of("parquet", (s, blockFactory) -> new ParquetFormatReader(blockFactory));

x-pack/plugin/esql-datasource-s3/src/main/java/org/elasticsearch/xpack/esql/datasource/s3/S3DataSourcePlugin.java

@@ -14,13 +14,19 @@
 import org.elasticsearch.xpack.esql.datasources.spi.StorageProviderFactory;
 
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Data source plugin providing S3 storage support for ESQL.
  * Supports s3://, s3a://, and s3n:// URI schemes.
  */
 public class S3DataSourcePlugin extends Plugin implements DataSourcePlugin {
 
+    @Override
+    public Set<String> supportedSchemes() {
+        return Set.of("s3", "s3a", "s3n");
+    }
+
     @Override
     public Map<String, StorageProviderFactory> storageProviders(Settings settings) {
         StorageProviderFactory s3Factory = new StorageProviderFactory() {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/datasources/DataSourceCapabilities.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.datasources;
+
+import org.elasticsearch.xpack.esql.datasources.spi.DataSourcePlugin;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+
+/**
+ * Aggregated capability declarations from all registered {@link DataSourcePlugin} instances.
+ * Built once at startup from cheap SPI calls; does not store plugin references.
+ */
+public record DataSourceCapabilities(Set<String> schemes, Set<String> formats, Set<String> extensions, Set<String> catalogs) {
+
+    public boolean supportsScheme(String s) {
+        return s != null && schemes.contains(s.toLowerCase(Locale.ROOT));
+    }
+
+    public boolean supportsFormat(String f) {
+        return f != null && formats.contains(f.toLowerCase(Locale.ROOT));
+    }
+
+    public boolean supportsExtension(String ext) {
+        if (ext == null) {
+            return false;
+        }
+        String normalized = ext.toLowerCase(Locale.ROOT);
+        if (normalized.startsWith(".") == false) {
+            normalized = "." + normalized;
+        }
+        return extensions.contains(normalized);
+    }
+
+    public boolean supportsCatalog(String c) {
+        return c != null && catalogs.contains(c.toLowerCase(Locale.ROOT));
+    }
+
+    public String supportedSchemesString() {
+        return String.join(", ", schemes);
+    }
+
+    public static DataSourceCapabilities build(List<DataSourcePlugin> plugins) {
+        Set<String> allSchemes = new LinkedHashSet<>();
+        Set<String> allFormats = new LinkedHashSet<>();
+        Set<String> allExtensions = new LinkedHashSet<>();
+        Set<String> allCatalogs = new LinkedHashSet<>();
+
+        for (DataSourcePlugin plugin : plugins) {
+            for (String scheme : plugin.supportedSchemes()) {
+                allSchemes.add(scheme.toLowerCase(Locale.ROOT));
+            }
+            for (String scheme : plugin.supportedConnectorSchemes()) {
+                allSchemes.add(scheme.toLowerCase(Locale.ROOT));
+            }
+            for (String format : plugin.supportedFormats()) {
+                String lower = format.toLowerCase(Locale.ROOT);
+                if (allFormats.contains(lower)) {
+                    throw new IllegalArgumentException("Format reader for [" + format + "] is already registered");
+                }
+                allFormats.add(lower);
+            }
+            for (String ext : plugin.supportedExtensions()) {
+                String normalized = ext.toLowerCase(Locale.ROOT);
+                if (normalized.startsWith(".") == false) {
+                    normalized = "." + normalized;
+                }
+                allExtensions.add(normalized);
+            }
+            for (String catalog : plugin.supportedCatalogs()) {
+                String lower = catalog.toLowerCase(Locale.ROOT);
+                if (allCatalogs.contains(lower)) {
+                    throw new IllegalArgumentException("Source factory for type [" + catalog + "] is already registered");
+                }
+                allCatalogs.add(lower);
+            }
+        }
+
+        return new DataSourceCapabilities(
+            Collections.unmodifiableSet(allSchemes),
+            Collections.unmodifiableSet(allFormats),
+            Collections.unmodifiableSet(allExtensions),
+            Collections.unmodifiableSet(allCatalogs)
+        );
+    }
+}