coordinatorSplits = new ArrayList<>();
+ if (operatorFactoryRegistry != null) {
+ physicalPlan.forEachDown(FragmentExec.class, fragment -> fragment.fragment().forEachDown(ExternalRelation.class, external -> {
+ ExternalSourceExec tempExec = external.toPhysicalExec();
+ PhysicalPlan discovered = SplitDiscoveryPhase.resolveExternalSplits(tempExec, operatorFactoryRegistry.sourceFactories());
+ if (discovered instanceof ExternalSourceExec withSplits) {
+ if (withSplits.splits().isEmpty() == false) {
+ coordinatorSplits.addAll(withSplits.splits());
+ } else {
+ // Fallback: FileSplitProvider returns empty for FileSet.UNRESOLVED (single-file).
+ // Create a single split from the path so the operator can read.
+ String path = withSplits.sourcePath();
+ if (path != null && path.startsWith("file://")) {
+ try {
+ StoragePath storagePath = StoragePath.of(path);
+ Path fsPath = PathUtils.get(URI.create(path));
+ if (Files.exists(fsPath)) {
+ long fileLength = Files.size(fsPath);
+ coordinatorSplits.add(
+ new FileSplit(
+ withSplits.sourceType(),
+ storagePath,
+ 0,
+ fileLength,
+ ".csv",
+ withSplits.config() != null ? withSplits.config() : Map.of(),
+ Map.of()
+ )
+ );
+ }
+ } catch (Exception e) {
+ LOGGER.debug(
+ () -> org.elasticsearch.core.Strings.format(
+ "Fallback split creation failed for path [%s]; "
+ + "operator may still work with path directly when splits empty",
+ path
+ ),
+ e
+ );
+ }
+ }
+ }
+ }
+ }));
+ }
+ return coordinatorSplits;
+ }
+
+ /**
+ * Returns a template resolver that maps CSV dataset names (e.g. "employees") to file:// URLs
+ * for datasets in CSV_DATASET that have a data file. Unknown template names return null.
+ *
+ * Paths are resolved lazily on first use per template and cached. For jar resources, temp files
+ * use a file-specific prefix; if a file matching that prefix already exists, it is reused.
+ */
+ public static Function csvFileTemplateResolver() {
+ ConcurrentHashMap cache = new ConcurrentHashMap<>();
+ return templateName -> {
+ CsvTestsDataLoader.TestDataset dataset = CSV_DATASET.get(templateName);
+ if (dataset == null || dataset.dataFileName() == null) {
+ throw new IllegalArgumentException("Failed to resolve CSV path for dataset [" + templateName + "]");
+ }
+ return cache.computeIfAbsent(templateName, k -> {
+ try {
+ return resolveCsvFilePathLazy("/data/" + dataset.dataFileName());
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to resolve CSV path for dataset [" + templateName + "]", e);
+ }
+ });
+ };
+ }
+
+ private static String resolveCsvFilePathLazy(String resourcePath) throws IOException {
+ URL resource = CsvTestsDataLoader.class.getResource(resourcePath);
+ if (resource == null) {
+ throw new IllegalStateException("Cannot find resource " + resourcePath);
+ }
+ String protocol = resource.getProtocol();
+ if ("file".equals(protocol)) {
+ return normalizeFileUri(resource.toExternalForm());
+ }
+ if ("jar".equals(protocol)) {
+ String fileName = PathUtils.get(resourcePath).getFileName().toString();
+ String prefix = CsvTests.class.getCanonicalName() + "-" + fileName + "-";
+ Path tempDir = PathUtils.get(System.getProperty("java.io.tmpdir"));
+ try (var stream = Files.newDirectoryStream(tempDir, prefix + "*")) {
+ var iterator = stream.iterator();
+ if (iterator.hasNext()) {
+ Path existing = iterator.next();
+ return normalizeFileUri(existing.toUri().toURL().toExternalForm());
+ }
+ }
+ Path tempFile = createTempFile(tempDir, prefix, ".csv");
+ try (InputStream in = getResourceStream(resourcePath)) {
+ Files.copy(in, tempFile, StandardCopyOption.REPLACE_EXISTING);
+ }
+ return normalizeFileUri(tempFile.toUri().toURL().toExternalForm());
+ }
+ throw new IllegalStateException("Unsupported resource protocol: " + protocol);
+ }
+
+ private static Path createTempFile(Path tempDir, String prefix, String suffix) throws IOException {
+ Path tempFile = Files.createTempFile(tempDir, prefix, suffix);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ try {
+ Files.deleteIfExists(tempFile);
+ } catch (IOException ignored) {
+ // Best-effort cleanup on shutdown
+ }
+ }));
+ return tempFile;
+ }
+
+ private static String normalizeFileUri(String uri) {
+ if (uri != null && uri.startsWith("file:/") && uri.startsWith("file:///") == false) {
+ return "file://" + uri.substring(5);
+ }
+ return uri;
+ }
+
+ public static String substituteTemplates(String query, Function templateResolver) {
+ Matcher matcher = TEMPLATE_PATTERN.matcher(query);
+ StringBuilder result = new StringBuilder();
+ while (matcher.find()) {
+ String templateName = matcher.group(1);
+ String replacement = templateResolver.apply(templateName);
+ matcher.appendReplacement(result, Matcher.quoteReplacement(replacement != null ? replacement : matcher.group(0)));
+ }
+ matcher.appendTail(result);
+ return result.toString();
+ }
}