diff --git a/.gitignore b/.gitignore index 1f2fe4ab3a1a8..e71cc6bdebad8 100644 --- a/.gitignore +++ b/.gitignore @@ -24,7 +24,7 @@ benchmarks/build-eclipse-default/* server/bin/* server/build-eclipse-default/* test/framework/build-eclipse-default/* - +**/*.dylib # eclipse files .project .classpath diff --git a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsSearchBackendPlugin.java b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsSearchBackendPlugin.java index 373fcde77b75f..d782770736d8a 100644 --- a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsSearchBackendPlugin.java +++ b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsSearchBackendPlugin.java @@ -8,17 +8,16 @@ package org.opensearch.analytics.spi; -import org.opensearch.analytics.backend.EngineResultStream; -import org.opensearch.analytics.backend.ExecutionContext; -import org.opensearch.analytics.backend.SearchExecEngine; -import org.opensearch.index.engine.dataformat.DataFormat; - import java.util.Collections; -import java.util.List; import java.util.Set; /** * SPI extension point for back-end query engines for query planning and execution capabilities + * as needed by the {@link org.opensearch.analytics.exec.QueryPlanExecutor}. + *

+ * Storage format declarations ({@code getSupportedFormats()}) belong on + * {@link org.opensearch.plugins.SearchBackEndPlugin} — the planner accesses + * field storage via {@code FieldStorageResolver} which reads from the storage layer. * as needed by the {@link org.opensearch.analytics.exec.QueryPlanExecutor} * *

TODO: separate capability declaration (planner, coordinator) from execution engine factory @@ -28,21 +27,6 @@ */ public interface AnalyticsSearchBackendPlugin extends SearchExecEngineProvider { - /** Unique engine name (e.g., "lucene", "datafusion"). */ - String name(); - - /** - * {@inheritDoc} - * Temporary default — remove once SearchExecEngineProvider is separated from this interface. - */ - @Override - default SearchExecEngine createSearchExecEngine(ExecutionContext ctx) { - throw new UnsupportedOperationException("createSearchExecEngine not implemented for " + name()); - } - - /** Returns the data formats supported by this backend. */ - List getSupportedFormats(); - /** Filter capabilities scoped to operator, field type, and data format. */ default Set filterCapabilities() { return Collections.emptySet(); diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/jni/NativeBridge.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/jni/NativeBridge.java index db14b6d40ae4c..f522573bc60ee 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/jni/NativeBridge.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/jni/NativeBridge.java @@ -25,10 +25,44 @@ private NativeBridge() {} private static synchronized void loadNativeLibrary() { if (loaded) return; try { + // Try java.library.path first System.loadLibrary("opensearch_datafusion_jni"); loaded = true; } catch (UnsatisfiedLinkError e) { - throw new ExceptionInInitializerError("Failed to load native library opensearch_datafusion_jni: " + e.getMessage()); + // Fall back to loading from classpath resources + try { + loadFromResources(); + loaded = true; + } catch (Exception ex) { + throw new ExceptionInInitializerError( + "Failed to load native library opensearch_datafusion_jni: " + e.getMessage() + + ". Also failed to load from resources: " + ex.getMessage()); + } + } + } + + private static void loadFromResources() throws java.io.IOException { + String os = System.getProperty("os.name", "").toLowerCase(java.util.Locale.ROOT); + String libName; + if (os.contains("mac")) { + libName = "libopensearch_datafusion_jni.dylib"; + } else if (os.contains("win")) { + libName = "opensearch_datafusion_jni.dll"; + } else { + libName = "libopensearch_datafusion_jni.so"; + } + + String resourcePath = "/native/" + libName; + try (java.io.InputStream in = NativeBridge.class.getResourceAsStream(resourcePath)) { + if (in == null) { + throw new java.io.FileNotFoundException("Native library not found in resources: " + resourcePath); + } + java.io.File tempFile = java.io.File.createTempFile("opensearch_datafusion_jni", libName.substring(libName.lastIndexOf('.'))); + tempFile.deleteOnExit(); + try (java.io.OutputStream out = new java.io.FileOutputStream(tempFile)) { + in.transferTo(out); + } + System.load(tempFile.getAbsolutePath()); } } diff --git a/sandbox/plugins/analytics-backend-lucene/build.gradle b/sandbox/plugins/analytics-backend-lucene/build.gradle index 3ee3ae85860c0..355d996d4177c 100644 --- a/sandbox/plugins/analytics-backend-lucene/build.gradle +++ b/sandbox/plugins/analytics-backend-lucene/build.gradle @@ -11,15 +11,15 @@ apply plugin: 'opensearch.internal-cluster-test' opensearchplugin { description = 'OpenSearch plugin providing Lucene-based search execution engine' classname = 'org.opensearch.be.lucene.LuceneSearchEnginePlugin' + extendedPlugins = ['analytics-engine'] } dependencies { - // Shared types and SPI interfaces (EngineBridge, AnalyticsBackEndPlugin, etc.) - // Also provides calcite-core transitively via api. - api project(':sandbox:libs:analytics-framework') + // Provided at runtime by the parent analytics-engine plugin (via extendedPlugins). + compileOnly project(':sandbox:libs:analytics-framework') - implementation "org.apache.logging.log4j:log4j-api:${versions.log4j}" - implementation "org.apache.logging.log4j:log4j-core:${versions.log4j}" + compileOnly "org.apache.logging.log4j:log4j-api:${versions.log4j}" + compileOnly "org.apache.logging.log4j:log4j-core:${versions.log4j}" } test { diff --git a/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneDataFormat.java b/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneDataFormat.java new file mode 100644 index 0000000000000..efa2ef8420eac --- /dev/null +++ b/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneDataFormat.java @@ -0,0 +1,89 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.be.lucene; + +import org.opensearch.index.engine.dataformat.DataFormat; +import org.opensearch.index.engine.dataformat.FieldTypeCapabilities; +import org.opensearch.index.engine.dataformat.FieldTypeCapabilities.Capability; + +import java.util.Set; + +/** + * Lucene data format capabilities. Declares what Lucene provides for each field type: + *

+ */ +public class LuceneDataFormat extends DataFormat { + + public static final LuceneDataFormat INSTANCE = new LuceneDataFormat(); + + private static final Set DOC_VALUES_INDEX_STORED = Set.of( + Capability.COLUMNAR_STORAGE, Capability.FULL_TEXT_SEARCH, Capability.STORED_FIELDS + ); + + private static final Set DOC_VALUES_POINT_STORED = Set.of( + Capability.COLUMNAR_STORAGE, Capability.POINT_RANGE, Capability.STORED_FIELDS + ); + + private static final Set FULL_TEXT_STORED = Set.of( + Capability.FULL_TEXT_SEARCH, Capability.STORED_FIELDS + ); + + private static final Set DOC_VALUES_STORED = Set.of( + Capability.COLUMNAR_STORAGE, Capability.STORED_FIELDS + ); + + private static final Set SUPPORTED_FIELDS = Set.of( + // String types + new FieldTypeCapabilities("keyword", DOC_VALUES_INDEX_STORED), + new FieldTypeCapabilities("text", FULL_TEXT_STORED), + + // Numeric types — all have doc values + point range + new FieldTypeCapabilities("integer", DOC_VALUES_POINT_STORED), + new FieldTypeCapabilities("long", DOC_VALUES_POINT_STORED), + new FieldTypeCapabilities("short", DOC_VALUES_POINT_STORED), + new FieldTypeCapabilities("byte", DOC_VALUES_POINT_STORED), + new FieldTypeCapabilities("float", DOC_VALUES_POINT_STORED), + new FieldTypeCapabilities("double", DOC_VALUES_POINT_STORED), + new FieldTypeCapabilities("half_float", DOC_VALUES_POINT_STORED), + new FieldTypeCapabilities("scaled_float", DOC_VALUES_POINT_STORED), + + // Date + new FieldTypeCapabilities("date", DOC_VALUES_POINT_STORED), + new FieldTypeCapabilities("date_nanos", DOC_VALUES_POINT_STORED), + + // Boolean + new FieldTypeCapabilities("boolean", DOC_VALUES_STORED), + + // IP + new FieldTypeCapabilities("ip", Set.of(Capability.COLUMNAR_STORAGE, Capability.POINT_RANGE, Capability.STORED_FIELDS)) + ); + + @Override + public String name() { + return "lucene"; + } + + @Override + public long priority() { + return 100; + } + + @Override + public Set supportedFields() { + return SUPPORTED_FIELDS; + } +} diff --git a/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearchEnginePlugin.java b/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearchEnginePlugin.java index 73fe9dda38bb7..8659c61b731d5 100644 --- a/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearchEnginePlugin.java +++ b/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearchEnginePlugin.java @@ -9,9 +9,19 @@ package org.opensearch.be.lucene; import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.search.MatchAllDocsQuery; +import org.opensearch.analytics.backend.EngineResultStream; +import org.opensearch.analytics.backend.ExecutionContext; +import org.opensearch.analytics.backend.SearchExecEngine; +import org.opensearch.analytics.spi.AnalyticsSearchBackendPlugin; +import org.opensearch.analytics.spi.OperatorCapability; import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.index.IndexSettings; import org.opensearch.index.engine.dataformat.DataFormat; +import org.opensearch.index.engine.dataformat.DataFormatPlugin; +import org.opensearch.index.engine.dataformat.IndexingExecutionEngine; import org.opensearch.index.engine.exec.EngineReaderManager; +import org.opensearch.index.mapper.MapperService; import org.opensearch.index.shard.ShardPath; import org.opensearch.plugins.Plugin; import org.opensearch.plugins.SearchBackEndPlugin; @@ -20,14 +30,15 @@ import java.util.List; /** - * Plugin providing Lucene as an index filter or source provider. + * Plugin providing Lucene as both a storage backend ({@link SearchBackEndPlugin}) + * and an analytics execution backend ({@link AnalyticsSearchBackendPlugin}). * * @opensearch.experimental */ @ExperimentalApi -public class LuceneSearchEnginePlugin extends Plugin implements SearchBackEndPlugin { +public class LuceneSearchEnginePlugin extends Plugin + implements SearchBackEndPlugin, AnalyticsSearchBackendPlugin, DataFormatPlugin { - /** Creates a new LuceneSearchEnginePlugin. */ public LuceneSearchEnginePlugin() {} @Override @@ -35,6 +46,8 @@ public String name() { return "lucene-analytics-backend"; } + // ---- SearchBackEndPlugin (storage) ---- + @Override public EngineReaderManager createReaderManager(DataFormat format, ShardPath shardPath) throws IOException { return new LuceneReaderManager(format); @@ -42,6 +55,42 @@ public EngineReaderManager createReaderManager(DataFormat forma @Override public List getSupportedFormats() { - return List.of(); + return List.of(LuceneDataFormat.INSTANCE); + } + + // ---- DataFormatPlugin (format registration) ---- + + @Override + public DataFormat getDataFormat() { + return LuceneDataFormat.INSTANCE; + } + + @Override + public IndexingExecutionEngine indexingEngine(MapperService mapperService, ShardPath shardPath, IndexSettings indexSettings) { + // Lucene indexing is handled by OpenSearch core, not this plugin. + return null; + } + + // ---- AnalyticsSearchBackendPlugin (capabilities + execution) ---- + + @Override + public java.util.Set supportedOperators() { + return java.util.Set.of( + OperatorCapability.SCAN, + OperatorCapability.FILTER, + OperatorCapability.PROJECT, + OperatorCapability.SORT + ); + } + + @Override + public SearchExecEngine createSearchExecEngine(ExecutionContext ctx) { + try { + DirectoryReader reader = ctx.getReader().getReader(LuceneDataFormat.INSTANCE, DirectoryReader.class); + LuceneSearchContext luceneCtx = new LuceneSearchContext(ctx.getTask(), reader, new MatchAllDocsQuery()); + return new LuceneSearchExecEngine(luceneCtx); + } catch (IOException e) { + throw new RuntimeException("Failed to create Lucene search exec engine", e); + } } } diff --git a/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearchExecEngine.java b/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearchExecEngine.java new file mode 100644 index 0000000000000..b2c3520583ba3 --- /dev/null +++ b/sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearchExecEngine.java @@ -0,0 +1,51 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.be.lucene; + +import org.opensearch.analytics.backend.EngineResultStream; +import org.opensearch.analytics.backend.ExecutionContext; +import org.opensearch.analytics.backend.SearchExecEngine; +import org.opensearch.common.annotation.ExperimentalApi; + +import java.io.IOException; + +/** + * Lucene-backed search execution engine. + * + * @opensearch.experimental + */ +@ExperimentalApi +public class LuceneSearchExecEngine implements SearchExecEngine { + + private final LuceneSearchContext context; + + public LuceneSearchExecEngine(LuceneSearchContext context) { + this.context = context; + } + + @Override + public void prepare(ExecutionContext requestContext) { + // TODO: extract query from plan and set on context + } + + @Override + public EngineResultStream execute(ExecutionContext requestContext) throws IOException { + // TODO: execute via LuceneEngineSearcher and return result stream + return null; + } + + public LuceneSearchContext getContext() { + return context; + } + + @Override + public void close() throws IOException { + context.close(); + } +} diff --git a/sandbox/plugins/analytics-backend-lucene/src/main/resources/META-INF/services/org.opensearch.analytics.spi.AnalyticsSearchBackendPlugin b/sandbox/plugins/analytics-backend-lucene/src/main/resources/META-INF/services/org.opensearch.analytics.spi.AnalyticsSearchBackendPlugin new file mode 100644 index 0000000000000..53330f0ac02ef --- /dev/null +++ b/sandbox/plugins/analytics-backend-lucene/src/main/resources/META-INF/services/org.opensearch.analytics.spi.AnalyticsSearchBackendPlugin @@ -0,0 +1 @@ +org.opensearch.be.lucene.LuceneSearchEnginePlugin diff --git a/sandbox/plugins/analytics-backend-lucene/src/main/resources/META-INF/services/org.opensearch.plugins.SearchBackEndPlugin b/sandbox/plugins/analytics-backend-lucene/src/main/resources/META-INF/services/org.opensearch.plugins.SearchBackEndPlugin new file mode 100644 index 0000000000000..53330f0ac02ef --- /dev/null +++ b/sandbox/plugins/analytics-backend-lucene/src/main/resources/META-INF/services/org.opensearch.plugins.SearchBackEndPlugin @@ -0,0 +1 @@ +org.opensearch.be.lucene.LuceneSearchEnginePlugin diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java index 616ec205788c3..d556692cebf5a 100644 --- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java @@ -15,9 +15,12 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.opensearch.analytics.exec.DefaultPlanExecutor; +import org.opensearch.analytics.planner.CapabilityRegistry; import org.opensearch.analytics.exec.QueryPlanExecutor; +import org.opensearch.analytics.planner.FieldStorageResolver; import org.opensearch.analytics.schema.OpenSearchSchemaBuilder; import org.opensearch.analytics.spi.AnalyticsSearchBackendPlugin; +import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.cluster.metadata.IndexNameExpressionResolver; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.inject.Module; @@ -36,7 +39,10 @@ import java.util.ArrayList; import java.util.Collection; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; +import java.util.function.Function; import java.util.function.Supplier; /** @@ -55,12 +61,14 @@ public class AnalyticsPlugin extends Plugin implements ExtensiblePlugin { public AnalyticsPlugin() {} private final List backEnds = new ArrayList<>(); + private final List> storageBackends = new ArrayList<>(); private SqlOperatorTable operatorTable; - @SuppressWarnings("rawtypes") + @SuppressWarnings({"rawtypes", "unchecked"}) @Override public void loadExtensions(ExtensionLoader loader) { backEnds.addAll(loader.loadExtensions(AnalyticsSearchBackendPlugin.class)); + storageBackends.addAll((List) loader.loadExtensions(org.opensearch.plugins.SearchBackEndPlugin.class)); operatorTable = aggregateOperatorTables(); } @@ -79,7 +87,19 @@ public Collection createComponents( Supplier repositoriesServiceSupplier ) { DefaultEngineContext ctx = new DefaultEngineContext(clusterService, operatorTable); - DefaultPlanExecutor executor = new DefaultPlanExecutor(backEnds, null/* TODO: pass indices service */, clusterService); + + // Build scan format index and field storage factory from storage backends + Map> scanFormats = new LinkedHashMap<>(); + for (var sb : storageBackends) { + for (var format : sb.getSupportedFormats()) { + scanFormats.computeIfAbsent(format.name(), k -> new ArrayList<>()).add(sb.name()); + } + } + Function fieldStorageFactory = + (indexMetadata) -> new FieldStorageResolver(indexMetadata, storageBackends); + + CapabilityRegistry capabilityRegistry = new CapabilityRegistry(backEnds, fieldStorageFactory, scanFormats); + DefaultPlanExecutor executor = new DefaultPlanExecutor(backEnds, capabilityRegistry, clusterService); AnalyticsEngineService.setInstance(new AnalyticsEngineService(ctx, executor)); return List.of(executor, ctx); } @@ -102,7 +122,7 @@ private SqlOperatorTable aggregateOperatorTables() { /** * Default implementation of {@link EngineContext}. */ - static record DefaultEngineContext(ClusterService clusterService, SqlOperatorTable operatorTable) implements EngineContext { + record DefaultEngineContext(ClusterService clusterService, SqlOperatorTable operatorTable) implements EngineContext { @Override public SchemaPlus getSchema() { diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/DefaultPlanExecutor.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/DefaultPlanExecutor.java index e1d42f9f5a645..c87583f1e7722 100644 --- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/DefaultPlanExecutor.java +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/DefaultPlanExecutor.java @@ -9,131 +9,50 @@ package org.opensearch.analytics.exec; import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.core.TableScan; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import org.opensearch.action.search.SearchShardTask; -import org.opensearch.analytics.backend.EngineResultBatch; -import org.opensearch.analytics.backend.EngineResultStream; -import org.opensearch.analytics.backend.ExecutionContext; -import org.opensearch.analytics.backend.SearchExecEngine; import org.opensearch.analytics.planner.CapabilityRegistry; import org.opensearch.analytics.planner.PlannerContext; import org.opensearch.analytics.planner.PlannerImpl; import org.opensearch.analytics.spi.AnalyticsSearchBackendPlugin; import org.opensearch.cluster.service.ClusterService; -import org.opensearch.index.IndexService; -import org.opensearch.index.engine.DataFormatAwareEngine; -import org.opensearch.index.shard.IndexShard; -import org.opensearch.indices.IndicesService; import java.util.ArrayList; -import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Set; /** - * {@link QueryPlanExecutor} default implementation. - *

- * Acquires a composite reader, selects a {@link AnalyticsSearchBackendPlugin}, and - * delegates query execution to it. + * Coordinator-level plan executor. Plans the query via the capability-aware planner. + * Shard-level execution will be handled by a separate data-node transport action. */ public class DefaultPlanExecutor implements QueryPlanExecutor> { private static final Logger logger = LogManager.getLogger(DefaultPlanExecutor.class); private final Map backEnds; - private final IndicesService indicesService; + private final CapabilityRegistry capabilityRegistry; private final ClusterService clusterService; - /** - * Constructs a DefaultPlanExecutor. - * - * @param providers list of search execution engine providers - * @param indicesService service for accessing index shards - * @param clusterService service for accessing cluster state - */ - public DefaultPlanExecutor(List providers, IndicesService indicesService, ClusterService clusterService) { + public DefaultPlanExecutor(List providers, + CapabilityRegistry capabilityRegistry, + ClusterService clusterService) { this.backEnds = new LinkedHashMap<>(); for (AnalyticsSearchBackendPlugin provider : providers) { this.backEnds.put(provider.name(), provider); } - this.indicesService = indicesService; + this.capabilityRegistry = capabilityRegistry; this.clusterService = clusterService; } @Override public Iterable execute(RelNode logicalFragment, Object context) { logicalFragment = PlannerImpl.createPlan(logicalFragment, - new PlannerContext( - new CapabilityRegistry(new ArrayList<>(backEnds.values())), - clusterService.state())); - String tableName = extractTableName(logicalFragment); - AnalyticsSearchBackendPlugin provider = selectBackEnd(); - if (provider == null) { - return new ArrayList<>(); - } - - IndexShard shard = resolveShard(tableName); - DataFormatAwareEngine dataFormatAwareEngine = shard.getCompositeEngine(); - if (dataFormatAwareEngine == null) { - throw new IllegalStateException("No CompositeEngine on shard [" + shard.shardId() + "]"); - } - - SearchShardTask task = null; // TODO: init task - List rows = new ArrayList<>(); - try (var dataFormatAwareReader = dataFormatAwareEngine.acquireReader()) { - ExecutionContext ctx = new ExecutionContext(tableName, task, dataFormatAwareReader.get()); - try (SearchExecEngine engine = provider.createSearchExecEngine(ctx)) { - logger.info("[DefaultPlanExecutor] Executing via [{}]", provider.name()); - try (EngineResultStream resultStream = engine.execute(ctx)) { - Iterator batchIterator = resultStream.iterator(); - while (batchIterator.hasNext()) { - EngineResultBatch batch = batchIterator.next(); - List fieldNames = batch.getFieldNames(); - for (int row = 0; row < batch.getRowCount(); row++) { - Object[] rowValues = new Object[fieldNames.size()]; - for (int col = 0; col < fieldNames.size(); col++) { - rowValues[col] = batch.getFieldValue(fieldNames.get(col), row); - } - rows.add(rowValues); - } - } - } - } - } catch (Exception e) { - throw new RuntimeException("Execution failed for [" + provider.name() + "]", e); - } - return rows; - } + new PlannerContext(capabilityRegistry, clusterService.state())); - static String extractTableName(RelNode node) { - if (node instanceof TableScan) { - List qn = node.getTable().getQualifiedName(); - return qn.get(qn.size() - 1); - } - for (RelNode input : node.getInputs()) { - String name = extractTableName(input); - if (name != null) return name; - } - throw new IllegalArgumentException("No TableScan found in plan fragment"); - } + logger.info("[DefaultPlanExecutor] Planned:\n{}", logicalFragment.explain()); - private IndexShard resolveShard(String indexName) { - IndexService indexService = indicesService.indexService(clusterService.state().metadata().index(indexName).getIndex()); - if (indexService == null) throw new IllegalStateException("Index [" + indexName + "] not on this node"); - Set shardIds = indexService.shardIds(); - if (shardIds.isEmpty()) throw new IllegalStateException("No shards for [" + indexName + "]"); - return indexService.getShardOrNull(shardIds.iterator().next()); - } - - private AnalyticsSearchBackendPlugin selectBackEnd() { - if (backEnds.isEmpty()) { - logger.warn("No back-end plugins registered — queries will return empty results"); - return null; - } - // TODO: select based on data format available in the catalog snapshot - return backEnds.values().iterator().next(); + // TODO: dispatch to data-node transport action for shard-level execution + // For now, return empty results — the plan is logged for validation + return new ArrayList<>(); } } diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/CapabilityRegistry.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/CapabilityRegistry.java index 3994a86882ad7..48220ffd8204b 100644 --- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/CapabilityRegistry.java +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/CapabilityRegistry.java @@ -22,12 +22,14 @@ import org.opensearch.analytics.spi.ShuffleCapability; import org.opensearch.analytics.spi.WindowCapability; import org.opensearch.analytics.spi.WindowFunction; +import org.opensearch.cluster.metadata.IndexMetadata; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.function.Function; /** * Pre-indexed capability lookups for planner rules. Built once at plugin startup, @@ -54,9 +56,14 @@ public class CapabilityRegistry { private final Map> scanFormatIndex = new HashMap<>(); // backendName → ShuffleCapabilities private final Map> shuffleCapabilities = new HashMap<>(); + private final Function fieldStorageFactory; - public CapabilityRegistry(List backends) { + public CapabilityRegistry(List backends, + Function fieldStorageFactory, + Map> scanFormats) { this.backends = backends; + this.fieldStorageFactory = fieldStorageFactory; + this.scanFormatIndex.putAll(scanFormats); for (AnalyticsSearchBackendPlugin backend : backends) { String name = backend.name(); @@ -114,12 +121,7 @@ public CapabilityRegistry(List backends) { cap.formats(), name); } - // Scan format index - if (backend.supportedOperators().contains(OperatorCapability.SCAN)) { - for (var format : backend.getSupportedFormats()) { - scanFormatIndex.computeIfAbsent(format.name(), k -> new ArrayList<>()).add(name); - } - } + // Scan format index — populated later via setStorageBackends() // Shuffle capabilities if (!backend.supportedShuffleCapabilities().isEmpty()) { @@ -206,11 +208,16 @@ public Set getShuffleCapabilities(String backendName) { return shuffleCapabilities.getOrDefault(backendName, Set.of()); } - /** Returns the backend list for FieldStorageResolver construction. */ + /** Returns the analytics backends. */ public List getBackends() { return backends; } + /** Builds a FieldStorageResolver for the given index. */ + public FieldStorageResolver resolveFieldStorage(IndexMetadata indexMetadata) { + return fieldStorageFactory.apply(indexMetadata); + } + /** All backends that support this filter operator on this field type, any format. */ public List filterBackendsAnyFormat(FilterOperator operator, FieldType fieldType) { return allBackends(filterIndex.getOrDefault(new FilterKey(operator, fieldType), Map.of())); diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/FieldStorageResolver.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/FieldStorageResolver.java index 4858ce1d788c8..5f4c86523c0cf 100644 --- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/FieldStorageResolver.java +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/FieldStorageResolver.java @@ -8,13 +8,13 @@ package org.opensearch.analytics.planner; -import org.opensearch.analytics.spi.AnalyticsSearchBackendPlugin; import org.opensearch.analytics.spi.FieldType; import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.cluster.metadata.MappingMetadata; import org.opensearch.index.engine.dataformat.DataFormat; import org.opensearch.index.engine.dataformat.FieldTypeCapabilities; import org.opensearch.index.engine.dataformat.FieldTypeCapabilities.Capability; +import org.opensearch.plugins.SearchBackEndPlugin; import java.util.ArrayList; import java.util.LinkedHashMap; @@ -35,7 +35,6 @@ public class FieldStorageResolver { private final Map fieldStorage; - private final List docValueFormats; /** * Production: resolves per-field storage from IndexMetadata and backend capabilities. @@ -56,7 +55,7 @@ public class FieldStorageResolver { * as a narrowing hint in the interim. */ @SuppressWarnings("unchecked") - public FieldStorageResolver(IndexMetadata indexMetadata, List backends) { + public FieldStorageResolver(IndexMetadata indexMetadata, List> backends) { String indexName = indexMetadata.getIndex().getName(); MappingMetadata mapping = indexMetadata.mapping(); @@ -82,7 +81,6 @@ public FieldStorageResolver(IndexMetadata indexMetadata, List fieldStorage) { this.fieldStorage = fieldStorage; - this.docValueFormats = computeDocValueFormats(fieldStorage); } /** Resolves storage info for the requested fields. */ @@ -107,31 +104,15 @@ public List resolve(List fieldNames) { return result; } - /** Returns all unique data formats that hold doc values across all fields. Precomputed at creation. */ - public List docValueFormats() { - return docValueFormats; - } - - private static List computeDocValueFormats(Map fieldStorage) { - List formats = new ArrayList<>(); - for (FieldStorageInfo info : fieldStorage.values()) { - for (String format : info.getDocValueFormats()) { - if (!formats.contains(format)) { - formats.add(format); - } - } - } - return formats; - } /** * Builds a lookup: formatName → fieldType → FieldTypeCapabilities * from all backends' DataFormats. */ private static Map> buildFormatCapabilities( - List backends) { + List> backends) { Map> result = new LinkedHashMap<>(); - for (AnalyticsSearchBackendPlugin backend : backends) { + for (SearchBackEndPlugin backend : backends) { for (DataFormat format : backend.getSupportedFormats()) { Map byFieldType = result.computeIfAbsent( format.name(), k -> new LinkedHashMap<>()); diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchTableScanRule.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchTableScanRule.java index 56bd6e87cabaf..0e3fb5aba5a9a 100644 --- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchTableScanRule.java +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rules/OpenSearchTableScanRule.java @@ -57,8 +57,7 @@ public void onMatch(RelOptRuleCall call) { String primaryFormat = indexMetadata.getSettings().get("index.composite.primary_data_format", "lucene"); CapabilityRegistry registry = context.getCapabilityRegistry(); - FieldStorageResolver fieldStorageResolver = new FieldStorageResolver(indexMetadata, - registry.getBackends()); + FieldStorageResolver fieldStorageResolver = registry.resolveFieldStorage(indexMetadata); // TODO : This expects the FrontEnds to attach the row type with all fields. // TODO : How will they attach if we perform the index resolution diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/DefaultPlanExecutorTests.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/DefaultPlanExecutorTests.java index 459a7f3786abd..f4c917047e82e 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/DefaultPlanExecutorTests.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/DefaultPlanExecutorTests.java @@ -21,19 +21,17 @@ import org.apache.calcite.rel.type.RelDataTypeFactory; import org.apache.calcite.rex.RexBuilder; import org.apache.calcite.sql.type.SqlTypeName; +import org.opensearch.analytics.planner.CapabilityRegistry; import org.opensearch.analytics.backend.EngineResultBatch; import org.opensearch.analytics.backend.EngineResultStream; import org.opensearch.analytics.backend.ExecutionContext; import org.opensearch.analytics.backend.SearchExecEngine; import org.opensearch.analytics.spi.AnalyticsSearchBackendPlugin; -import org.opensearch.analytics.spi.OperatorCapability; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.metadata.IndexMetadata; -import org.opensearch.cluster.metadata.MappingMetadata; import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.concurrent.GatedCloseable; -import org.opensearch.common.settings.Settings; import org.opensearch.core.index.Index; import org.opensearch.index.IndexService; import org.opensearch.index.engine.DataFormatAwareEngine; @@ -78,92 +76,9 @@ public void setUp() throws Exception { cluster = RelOptCluster.create(planner, rexBuilder); } - /** - * extractTableName returns the table name from a TableScan node. - */ - public void testExtractTableNameFromTableScan() { - RelOptTable table = mockTable("schema", "my_index"); - TableScan scan = new StubTableScan(cluster, cluster.traitSet(), table); - assertEquals("my_index", DefaultPlanExecutor.extractTableName(scan)); - } - - /** - * extractTableName throws when no TableScan is found. - */ - public void testExtractTableNameThrowsForNonTableScan() { - RelNode stub = new StubRelNode(cluster, cluster.traitSet(), buildRowType(1)); - IllegalArgumentException ex = expectThrows(IllegalArgumentException.class, () -> DefaultPlanExecutor.extractTableName(stub)); - assertTrue(ex.getMessage().contains("No TableScan found")); - } - - /** - * End-to-end: write file sets → catalog snapshot → DataFormatAwareEngine → - * DefaultPlanExecutor.execute() with mock backend returns rows via EngineResultStream. - */ - public void testEndToEndExecuteWithMockBackend() throws IOException { - MockDataFormat format = new MockDataFormat(); - Path dir = createTempDir(); - - WriterFileSet fs1 = WriterFileSet.builder().directory(dir).writerGeneration(1L).addFile("gen1.parquet").addNumRows(2).build(); - WriterFileSet fs2 = WriterFileSet.builder().directory(dir).writerGeneration(2L).addFile("gen2.parquet").addNumRows(1).build(); - - Segment seg1 = Segment.builder(0L).addSearchableFiles(format, fs1).build(); - Segment seg2 = Segment.builder(1L).addSearchableFiles(format, fs2).build(); - - CatalogSnapshotManager snapshotManager = new CatalogSnapshotManager(1L, 1L, 0L, List.of(seg1, seg2), 2L, Map.of()); - - MockReaderManager readerManager = new MockReaderManager(format.name()); - try (GatedCloseable ref = snapshotManager.acquireSnapshot()) { - readerManager.afterRefresh(true, ref.get()); - } - - DataFormatAwareEngine engine = new DataFormatAwareEngine(Map.of(format, readerManager), snapshotManager); - - // Mock shard + cluster wiring - IndexShard shard = mock(IndexShard.class); - when(shard.getCompositeEngine()).thenReturn(engine); - - Index index = new Index("my_index", "uuid"); - IndexMetadata indexMetadata = mock(IndexMetadata.class); - when(indexMetadata.getIndex()).thenReturn(index); - when(indexMetadata.getSettings()).thenReturn( - Settings.builder().put("index.composite.primary_data_format", "mock-columnar").build() - ); - when(indexMetadata.getNumberOfShards()).thenReturn(1); - MappingMetadata mappingMetadata = mock(MappingMetadata.class); - when(mappingMetadata.sourceAsMap()).thenReturn( - Map.of("properties", Map.of("field_0", Map.of("type", "integer"))) - ); - when(indexMetadata.mapping()).thenReturn(mappingMetadata); - - Metadata metadata = mock(Metadata.class); - when(metadata.index("my_index")).thenReturn(indexMetadata); - - ClusterState clusterState = mock(ClusterState.class); - when(clusterState.metadata()).thenReturn(metadata); - - ClusterService clusterService = mock(ClusterService.class); - when(clusterService.state()).thenReturn(clusterState); - - IndexService indexService = mock(IndexService.class); - when(indexService.shardIds()).thenReturn(Set.of(0)); - when(indexService.getShardOrNull(0)).thenReturn(shard); - - IndicesService indicesService = mock(IndicesService.class); - when(indicesService.indexService(index)).thenReturn(indexService); - - MockBackendPlugin backendPlugin = new MockBackendPlugin(format); - DefaultPlanExecutor executor = new DefaultPlanExecutor(List.of(backendPlugin), indicesService, clusterService); - - RelOptTable table = mockTable("my_index"); - TableScan scan = new StubTableScan(cluster, cluster.traitSet(), table); - - Iterable results = executor.execute(scan, new Object()); - List rows = new ArrayList<>(); - results.forEach(rows::add); - - assertEquals(3, rows.size()); - } + // E2E execution test removed — DefaultPlanExecutor is coordinator-only now. + // Planning is tested via planner rule tests. Shard execution will be tested + // via the data-node transport action. private RelOptTable mockTable(String... qualifiedName) { RelOptTable table = mock(RelOptTable.class); @@ -410,16 +325,6 @@ public String name() { return "mock-backend"; } - @Override - public List getSupportedFormats() { - return List.of(format); - } - - @Override - public Set supportedOperators() { - return Set.of(OperatorCapability.SCAN, OperatorCapability.FILTER); - } - @Override public SearchExecEngine createSearchExecEngine(ExecutionContext ctx) { Object reader = ctx.getReader().reader(format); diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/BasePlannerRulesTests.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/BasePlannerRulesTests.java index 2104bada7ba7b..8e00baa9ce69d 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/BasePlannerRulesTests.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/BasePlannerRulesTests.java @@ -8,6 +8,12 @@ package org.opensearch.analytics.planner; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + import org.apache.calcite.jdbc.JavaTypeFactoryImpl; import org.apache.calcite.plan.RelOptCluster; import org.apache.calcite.plan.RelOptTable; @@ -30,11 +36,18 @@ import org.opensearch.cluster.metadata.MappingMetadata; import org.opensearch.cluster.metadata.Metadata; import org.opensearch.common.settings.Settings; +import org.opensearch.core.index.Index; +import org.opensearch.index.engine.dataformat.DataFormat; +import org.opensearch.index.engine.dataformat.FieldTypeCapabilities; +import org.opensearch.index.engine.exec.EngineReaderManager; +import org.opensearch.index.shard.ShardPath; +import org.opensearch.plugins.SearchBackEndPlugin; import org.opensearch.test.OpenSearchTestCase; import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.function.Function; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -101,7 +114,7 @@ protected PlannerContext buildContext(String primaryFormat, int shardCount, when(mappingMetadata.sourceAsMap()).thenReturn(mappingSource); IndexMetadata indexMetadata = mock(IndexMetadata.class); - when(indexMetadata.getIndex()).thenReturn(new org.opensearch.core.index.Index("test_index", "uuid")); + when(indexMetadata.getIndex()).thenReturn(new Index("test_index", "uuid")); when(indexMetadata.getSettings()).thenReturn( Settings.builder().put("index.composite.primary_data_format", primaryFormat).build() ); @@ -114,7 +127,30 @@ protected PlannerContext buildContext(String primaryFormat, int shardCount, ClusterState clusterState = mock(ClusterState.class); when(clusterState.metadata()).thenReturn(metadata); - return new PlannerContext(new CapabilityRegistry(backends), clusterState); + // Build scan format index from backend names → format names + Map> scanFormats = new LinkedHashMap<>(); + for (var backend : backends) { + if (backend.name().contains("lucene")) { + scanFormats.computeIfAbsent(MockLuceneBackend.LUCENE_DATA_FORMAT, k -> new ArrayList<>()).add(backend.name()); + } else if (backend.name().contains("parquet")) { + scanFormats.computeIfAbsent(MockDataFusionBackend.PARQUET_DATA_FORMAT, k -> new ArrayList<>()).add(backend.name()); + } + } + + // Build FieldStorageResolver factory — mock backends implement both interfaces + List> storageBackends = new ArrayList<>(); + for (var backend : backends) { + if (backend instanceof SearchBackEndPlugin sb) { + storageBackends.add(sb); + } + } + List> finalStorageBackends = storageBackends; + Function fieldStorageFactory = idx -> + new FieldStorageResolver(idx, finalStorageBackends); + + return new PlannerContext( + new CapabilityRegistry(backends, fieldStorageFactory, scanFormats), + clusterState); } // ---- Table builders ---- @@ -177,4 +213,77 @@ protected static class StubTableScan extends TableScan { super(cluster, traitSet, List.of(), table); } } + + /** Minimal SearchBackEndPlugin for test FieldStorageResolver construction. */ + static class MockStorageBackend implements SearchBackEndPlugin { + private final String formatName; + private final Set fieldCaps; + + MockStorageBackend(String formatName, Set fieldCaps) { + this.formatName = formatName; + this.fieldCaps = fieldCaps; + } + + /** Lucene: POINT_RANGE + STORED_FIELDS for numerics/dates, FULL_TEXT_SEARCH + STORED_FIELDS for text/keyword */ + static MockStorageBackend lucene() { + var C = FieldTypeCapabilities.Capability.class; + return new MockStorageBackend(MockLuceneBackend.LUCENE_DATA_FORMAT, Set.of( + new FieldTypeCapabilities("integer", + Set.of(FieldTypeCapabilities.Capability.POINT_RANGE, + FieldTypeCapabilities.Capability.STORED_FIELDS)), + new FieldTypeCapabilities("long", + Set.of(FieldTypeCapabilities.Capability.POINT_RANGE, + FieldTypeCapabilities.Capability.STORED_FIELDS)), + new FieldTypeCapabilities("keyword", + Set.of(FieldTypeCapabilities.Capability.FULL_TEXT_SEARCH, + FieldTypeCapabilities.Capability.STORED_FIELDS)), + new FieldTypeCapabilities("text", + Set.of(FieldTypeCapabilities.Capability.FULL_TEXT_SEARCH, + FieldTypeCapabilities.Capability.STORED_FIELDS)), + new FieldTypeCapabilities("boolean", + Set.of(FieldTypeCapabilities.Capability.STORED_FIELDS)), + new FieldTypeCapabilities("date", + Set.of(FieldTypeCapabilities.Capability.POINT_RANGE, + FieldTypeCapabilities.Capability.STORED_FIELDS)) + )); + } + + /** Parquet/DataFusion: COLUMNAR_STORAGE for all types */ + static MockStorageBackend parquet() { + return new MockStorageBackend(MockDataFusionBackend.PARQUET_DATA_FORMAT, Set.of( + new FieldTypeCapabilities("integer", + Set.of(FieldTypeCapabilities.Capability.COLUMNAR_STORAGE)), + new FieldTypeCapabilities("long", + Set.of(FieldTypeCapabilities.Capability.COLUMNAR_STORAGE)), + new FieldTypeCapabilities("keyword", + Set.of(FieldTypeCapabilities.Capability.COLUMNAR_STORAGE)), + new FieldTypeCapabilities("text", + Set.of(FieldTypeCapabilities.Capability.COLUMNAR_STORAGE)), + new FieldTypeCapabilities("boolean", + Set.of(FieldTypeCapabilities.Capability.COLUMNAR_STORAGE)), + new FieldTypeCapabilities("date", + Set.of(FieldTypeCapabilities.Capability.COLUMNAR_STORAGE)) + )); + } + + @Override public String name() { return formatName; } + + @Override + public List getSupportedFormats() { + return List.of(new DataFormat() { + @Override public String name() { return formatName; } + @Override public long priority() { return 0; } + @Override public Set supportedFields() { + return fieldCaps; + } + }); + } + + @Override + public EngineReaderManager createReaderManager( + DataFormat format, + ShardPath shardPath) { + return null; + } + } } diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockDataFusionBackend.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockDataFusionBackend.java index c7e36ef7789ec..51b4fc264be88 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockDataFusionBackend.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockDataFusionBackend.java @@ -20,6 +20,8 @@ import org.opensearch.analytics.spi.OperatorCapability; import org.opensearch.index.engine.dataformat.DataFormat; import org.opensearch.index.engine.dataformat.FieldTypeCapabilities; +import org.opensearch.index.engine.exec.EngineReaderManager; +import org.opensearch.index.shard.ShardPath; import static org.opensearch.index.engine.dataformat.FieldTypeCapabilities.Capability.COLUMNAR_STORAGE; @@ -32,7 +34,7 @@ * standard filter operators on NUMERIC/KEYWORD/DATE/BOOLEAN, and common aggregates. * No full-text support. */ -public class MockDataFusionBackend implements AnalyticsSearchBackendPlugin { +public class MockDataFusionBackend implements AnalyticsSearchBackendPlugin, org.opensearch.plugins.SearchBackEndPlugin { public static final String NAME = "mock-parquet"; public static final String PARQUET_DATA_FORMAT = "parquet"; @@ -92,6 +94,12 @@ public class MockDataFusionBackend implements AnalyticsSearchBackendPlugin { @Override public SearchExecEngine createSearchExecEngine(ExecutionContext ctx) { return null; } + @Override public Set supportedOperators() { return OPERATOR_CAPS; } + @Override public Set filterCapabilities() { return FILTER_CAPS; } + @Override public Set aggregateCapabilities() { return AGG_CAPS; } + + // ---- SearchBackEndPlugin (storage) ---- + @Override public List getSupportedFormats() { return List.of(new DataFormat() { @@ -110,7 +118,8 @@ public List getSupportedFormats() { }); } - @Override public Set supportedOperators() { return OPERATOR_CAPS; } - @Override public Set filterCapabilities() { return FILTER_CAPS; } - @Override public Set aggregateCapabilities() { return AGG_CAPS; } + @Override + public EngineReaderManager createReaderManager(DataFormat format, ShardPath shardPath) { + return null; + } } diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockLuceneBackend.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockLuceneBackend.java index 7828ae9f46f59..be88fc406370c 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockLuceneBackend.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockLuceneBackend.java @@ -18,6 +18,8 @@ import org.opensearch.analytics.spi.OperatorCapability; import org.opensearch.index.engine.dataformat.DataFormat; import org.opensearch.index.engine.dataformat.FieldTypeCapabilities; +import org.opensearch.index.engine.exec.EngineReaderManager; +import org.opensearch.index.shard.ShardPath; import static org.opensearch.index.engine.dataformat.FieldTypeCapabilities.Capability.FULL_TEXT_SEARCH; import static org.opensearch.index.engine.dataformat.FieldTypeCapabilities.Capability.POINT_RANGE; @@ -33,7 +35,7 @@ * only in parquet for default tests). Standard + full-text filter capabilities. * SCAN + FILTER only (no AGGREGATE). */ -public class MockLuceneBackend implements AnalyticsSearchBackendPlugin { +public class MockLuceneBackend implements AnalyticsSearchBackendPlugin, org.opensearch.plugins.SearchBackEndPlugin { public static final String NAME = "mock-lucene"; public static final String LUCENE_DATA_FORMAT = "lucene"; @@ -94,6 +96,11 @@ public class MockLuceneBackend implements AnalyticsSearchBackendPlugin { @Override public SearchExecEngine createSearchExecEngine(ExecutionContext ctx) { return null; } + @Override public Set supportedOperators() { return OPERATOR_CAPS; } + @Override public Set filterCapabilities() { return FILTER_CAPS; } + + // ---- SearchBackEndPlugin (storage) ---- + @Override public List getSupportedFormats() { return List.of(new DataFormat() { @@ -112,6 +119,8 @@ public List getSupportedFormats() { }); } - @Override public Set supportedOperators() { return OPERATOR_CAPS; } - @Override public Set filterCapabilities() { return FILTER_CAPS; } + @Override + public EngineReaderManager createReaderManager(DataFormat format, ShardPath shardPath) { + return null; + } } diff --git a/sandbox/scripts/setup-test-data.sh b/sandbox/scripts/setup-test-data.sh new file mode 100755 index 0000000000000..651c61c25a60a --- /dev/null +++ b/sandbox/scripts/setup-test-data.sh @@ -0,0 +1,81 @@ +#!/bin/bash +# Setup test indices and data for analytics engine development. +# Usage: ./sandbox/scripts/setup-test-data.sh [host:port] + +HOST="${1:-localhost:9200}" + +echo "=== Setting up test data on $HOST ===" + +# Delete existing indices (ignore errors if they don't exist) +curl -s -X DELETE "$HOST/parquet_logs" > /dev/null 2>&1 +curl -s -X DELETE "$HOST/parquet_metrics" > /dev/null 2>&1 + +# Create parquet_logs +echo -n "Creating parquet_logs... " +curl -s -X PUT "$HOST/parquet_logs" -H "Content-Type: application/json" -d '{ + "settings": { "number_of_shards": 1, "number_of_replicas": 0 }, + "mappings": { + "properties": { + "ts": { "type": "date" }, + "status": { "type": "integer" }, + "message": { "type": "keyword" }, + "ip_addr": { "type": "keyword" } + } + } +}' | jq -r '.acknowledged // .error.reason' + +# Create parquet_metrics +echo -n "Creating parquet_metrics... " +curl -s -X PUT "$HOST/parquet_metrics" -H "Content-Type: application/json" -d '{ + "settings": { "number_of_shards": 1, "number_of_replicas": 0 }, + "mappings": { + "properties": { + "ts": { "type": "date" }, + "cpu": { "type": "double" }, + "memory": { "type": "double" }, + "host": { "type": "keyword" } + } + } +}' | jq -r '.acknowledged // .error.reason' + +# Insert parquet_logs data +echo -n "Inserting parquet_logs data... " +curl -s -X POST "$HOST/parquet_logs/_bulk?refresh=true" -H "Content-Type: application/x-ndjson" -d ' +{"index":{}} +{"ts":"2024-01-15T10:30:00Z","status":200,"message":"Request completed","ip_addr":"192.168.1.1"} +{"index":{}} +{"ts":"2024-01-15T10:31:00Z","status":200,"message":"Health check OK","ip_addr":"192.168.1.2"} +{"index":{}} +{"ts":"2024-01-15T10:32:00Z","status":500,"message":"Internal server error","ip_addr":"192.168.1.3"} +{"index":{}} +{"ts":"2024-01-15T10:33:00Z","status":200,"message":"Request completed","ip_addr":"192.168.1.4"} +{"index":{}} +{"ts":"2024-01-15T10:34:00Z","status":404,"message":"Not found","ip_addr":"192.168.1.5"} +' | jq -r '"items=\(.items | length), errors=\(.errors)"' + +# Insert parquet_metrics data +echo -n "Inserting parquet_metrics data... " +curl -s -X POST "$HOST/parquet_metrics/_bulk?refresh=true" -H "Content-Type: application/x-ndjson" -d ' +{"index":{}} +{"ts":"2024-01-15T10:30:00Z","cpu":75.5,"memory":8192.5,"host":"host-1"} +{"index":{}} +{"ts":"2024-01-15T10:31:00Z","cpu":82.3,"memory":7680.5,"host":"host-2"} +{"index":{}} +{"ts":"2024-01-15T10:32:00Z","cpu":45.1,"memory":6144.0,"host":"host-1"} +{"index":{}} +{"ts":"2024-01-15T10:33:00Z","cpu":91.7,"memory":7200.0,"host":"host-3"} +' | jq -r '"items=\(.items | length), errors=\(.errors)"' + +# Verify +echo "" +echo "=== Verification ===" +echo -n "parquet_logs count: " +curl -s "$HOST/parquet_logs/_count" | jq -r '.count' +echo -n "parquet_metrics count: " +curl -s "$HOST/parquet_metrics/_count" | jq -r '.count' + +echo "" +echo "=== Sample queries ===" +echo 'curl -X POST "'$HOST'/_plugins/_ppl" -H "Content-Type: application/json" -d '"'"'{"query": "source = parquet_logs | fields ts, status, message"}'"'"'' +echo 'curl -X POST "'$HOST'/_plugins/_ppl" -H "Content-Type: application/json" -d '"'"'{"query": "source = parquet_logs | where status = 200 | stats count() as cnt"}'"'"'' +echo 'curl -X POST "'$HOST'/_plugins/_ppl" -H "Content-Type: application/json" -d '"'"'{"query": "source = parquet_metrics | stats avg(cpu) as avg_cpu by host"}'"'"''