Skip to content

Commit be1b5f1

Browse files
toepkerd-zztoepkerd-zz
committed
feat: Add REST endpoint to extract indices from PPL query
Adds POST /_plugins/_alerting/_ppl/indices endpoint that accepts a PPL query string and returns the list of indices it searches. Uses PPLSyntaxParser + AstBuilder to parse PPL into an UnresolvedPlan AST, then walks the tree with AbstractNodeVisitor to find Relation nodes and extract their QualifiedNames (index names). Includes PPL_INDEX_EXTRACT logging at each step for debugging.
1 parent b7dcc1c commit be1b5f1

9 files changed

Lines changed: 255 additions & 2 deletions

File tree

‎alerting/build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -248,6 +248,7 @@ task integTest(type: RestIntegTestTask) {
248248
}
249249
tasks.named("check").configure { dependsOn(integTest) }
250250
Zip bundle = (Zip) project.getTasks().getByName("bundlePlugin");
251+
bundle.exclude('**/antlr4-runtime*')
251252
bundle.rename('annotations-2\\.(.+)\\.jar', 'aws-sdk-annotations-2.$1.jar')
252253
integTest.dependsOn(bundle)
253254
integTest.getClusters().forEach{c -> c.plugin(project.getObjects().fileProperty().value(bundle.getArchiveFile()))}

‎alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ package org.opensearch.alerting
88
import org.opensearch.action.ActionRequest
99
import org.opensearch.alerting.action.ExecuteMonitorAction
1010
import org.opensearch.alerting.action.ExecuteWorkflowAction
11+
import org.opensearch.alerting.action.ExtractPPLIndicesAction
1112
import org.opensearch.alerting.action.GetDestinationsAction
1213
import org.opensearch.alerting.action.GetEmailAccountAction
1314
import org.opensearch.alerting.action.GetEmailGroupAction
@@ -35,6 +36,7 @@ import org.opensearch.alerting.resthandler.RestDeleteMonitorAction
3536
import org.opensearch.alerting.resthandler.RestDeleteWorkflowAction
3637
import org.opensearch.alerting.resthandler.RestExecuteMonitorAction
3738
import org.opensearch.alerting.resthandler.RestExecuteWorkflowAction
39+
import org.opensearch.alerting.resthandler.RestExtractPPLIndicesAction
3840
import org.opensearch.alerting.resthandler.RestGetAlertsAction
3941
import org.opensearch.alerting.resthandler.RestGetDestinationsAction
4042
import org.opensearch.alerting.resthandler.RestGetEmailAccountAction
@@ -75,6 +77,7 @@ import org.opensearch.alerting.transport.TransportDeleteWorkflowAction
7577
import org.opensearch.alerting.transport.TransportDocLevelMonitorFanOutAction
7678
import org.opensearch.alerting.transport.TransportExecuteMonitorAction
7779
import org.opensearch.alerting.transport.TransportExecuteWorkflowAction
80+
import org.opensearch.alerting.transport.TransportExtractPPLIndicesAction
7881
import org.opensearch.alerting.transport.TransportGetAlertsAction
7982
import org.opensearch.alerting.transport.TransportGetDestinationsAction
8083
import org.opensearch.alerting.transport.TransportGetEmailAccountAction
@@ -157,7 +160,6 @@ import org.opensearch.threadpool.ThreadPool
157160
import org.opensearch.transport.client.Client
158161
import org.opensearch.watcher.ResourceWatcherService
159162
import java.util.function.Supplier
160-
161163
/**
162164
* Entry point of the OpenDistro for Elasticsearch alerting plugin
163165
* This class initializes the [RestGetMonitorAction], [RestDeleteMonitorAction], [RestIndexMonitorAction] rest handlers.
@@ -236,6 +238,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
236238
RestGetWorkflowAction(),
237239
RestDeleteWorkflowAction(),
238240
RestGetRemoteIndexesAction(),
241+
RestExtractPPLIndicesAction(),
239242
RestIndexAlertingCommentAction(),
240243
RestSearchAlertingCommentAction(),
241244
RestDeleteAlertingCommentAction(),
@@ -270,6 +273,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
270273
ActionPlugin.ActionHandler(AlertingActions.DELETE_COMMENT_ACTION_TYPE, TransportDeleteAlertingCommentAction::class.java),
271274
ActionPlugin.ActionHandler(ExecuteWorkflowAction.INSTANCE, TransportExecuteWorkflowAction::class.java),
272275
ActionPlugin.ActionHandler(GetRemoteIndexesAction.INSTANCE, TransportGetRemoteIndexesAction::class.java),
276+
ActionPlugin.ActionHandler(ExtractPPLIndicesAction.INSTANCE, TransportExtractPPLIndicesAction::class.java),
273277
ActionPlugin.ActionHandler(DocLevelMonitorFanOutAction.INSTANCE, TransportDocLevelMonitorFanOutAction::class.java)
274278
)
275279
}

‎alerting/src/main/kotlin/org/opensearch/alerting/action/ExtractPPLIndicesAction.kt‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package org.opensearch.alerting.action
7+
8+
import org.opensearch.action.ActionType
9+
10+
class ExtractPPLIndicesAction private constructor() : ActionType<ExtractPPLIndicesResponse>(NAME, ::ExtractPPLIndicesResponse) {
11+
companion object {
12+
val INSTANCE = ExtractPPLIndicesAction()
13+
const val NAME = "cluster:admin/opendistro/alerting/ppl/extract_indices"
14+
}
15+
}

‎alerting/src/main/kotlin/org/opensearch/alerting/action/ExtractPPLIndicesRequest.kt‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package org.opensearch.alerting.action
7+
8+
import org.opensearch.action.ActionRequest
9+
import org.opensearch.action.ActionRequestValidationException
10+
import org.opensearch.core.common.io.stream.StreamInput
11+
import org.opensearch.core.common.io.stream.StreamOutput
12+
import java.io.IOException
13+
14+
class ExtractPPLIndicesRequest : ActionRequest {
15+
val query: String
16+
17+
constructor(query: String) : super() {
18+
this.query = query
19+
}
20+
21+
@Throws(IOException::class)
22+
constructor(sin: StreamInput) : super(sin) {
23+
query = sin.readString()
24+
}
25+
26+
override fun validate(): ActionRequestValidationException? = null
27+
28+
@Throws(IOException::class)
29+
override fun writeTo(out: StreamOutput) {
30+
super.writeTo(out)
31+
out.writeString(query)
32+
}
33+
}

‎alerting/src/main/kotlin/org/opensearch/alerting/action/ExtractPPLIndicesResponse.kt‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package org.opensearch.alerting.action
7+
8+
import org.opensearch.core.action.ActionResponse
9+
import org.opensearch.core.common.io.stream.StreamInput
10+
import org.opensearch.core.common.io.stream.StreamOutput
11+
import org.opensearch.core.xcontent.ToXContent
12+
import org.opensearch.core.xcontent.ToXContentObject
13+
import org.opensearch.core.xcontent.XContentBuilder
14+
import java.io.IOException
15+
16+
class ExtractPPLIndicesResponse : ActionResponse, ToXContentObject {
17+
val indices: List<String>
18+
19+
constructor(indices: List<String>) : super() {
20+
this.indices = indices
21+
}
22+
23+
@Throws(IOException::class)
24+
constructor(sin: StreamInput) : super(sin) {
25+
indices = sin.readStringList()
26+
}
27+
28+
@Throws(IOException::class)
29+
override fun writeTo(out: StreamOutput) {
30+
out.writeStringCollection(indices)
31+
}
32+
33+
override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
34+
builder.startObject()
35+
builder.field("indices", indices)
36+
builder.endObject()
37+
return builder
38+
}
39+
}

‎alerting/src/main/kotlin/org/opensearch/alerting/resthandler/RestExtractPPLIndicesAction.kt‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package org.opensearch.alerting.resthandler
7+
8+
import org.apache.logging.log4j.LogManager
9+
import org.opensearch.alerting.action.ExtractPPLIndicesAction
10+
import org.opensearch.alerting.action.ExtractPPLIndicesRequest
11+
import org.opensearch.rest.BaseRestHandler
12+
import org.opensearch.rest.RestHandler
13+
import org.opensearch.rest.RestRequest
14+
import org.opensearch.rest.action.RestToXContentListener
15+
import org.opensearch.transport.client.node.NodeClient
16+
17+
private val log = LogManager.getLogger(RestExtractPPLIndicesAction::class.java)
18+
19+
class RestExtractPPLIndicesAction : BaseRestHandler() {
20+
companion object {
21+
val ROUTE = "/_plugins/_alerting/ppl/_extract_indices"
22+
}
23+
24+
override fun getName(): String = "extract_ppl_indices_action"
25+
26+
override fun routes(): List<RestHandler.Route> {
27+
return listOf(
28+
RestHandler.Route(RestRequest.Method.POST, ROUTE)
29+
)
30+
}
31+
32+
override fun prepareRequest(
33+
request: RestRequest,
34+
client: NodeClient
35+
): RestChannelConsumer {
36+
val parser = request.contentParser()
37+
var query = ""
38+
while (parser.nextToken() != null) {
39+
if (parser.currentName() == "query") {
40+
parser.nextToken()
41+
query = parser.text()
42+
}
43+
}
44+
log.info("PPL_INDEX_EXTRACT: REST received query=$query")
45+
return RestChannelConsumer { channel ->
46+
client.execute(
47+
ExtractPPLIndicesAction.INSTANCE,
48+
ExtractPPLIndicesRequest(query),
49+
RestToXContentListener(channel)
50+
)
51+
}
52+
}
53+
}

‎alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportExtractPPLIndicesAction.kt‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package org.opensearch.alerting.transport
7+
8+
import org.apache.logging.log4j.LogManager
9+
import org.opensearch.action.support.ActionFilters
10+
import org.opensearch.action.support.HandledTransportAction
11+
import org.opensearch.alerting.action.ExtractPPLIndicesAction
12+
import org.opensearch.alerting.action.ExtractPPLIndicesRequest
13+
import org.opensearch.alerting.action.ExtractPPLIndicesResponse
14+
import org.opensearch.common.inject.Inject
15+
import org.opensearch.core.action.ActionListener
16+
import org.opensearch.sql.ast.AbstractNodeVisitor
17+
import org.opensearch.sql.ast.Node
18+
import org.opensearch.sql.ast.expression.QualifiedName
19+
import org.opensearch.sql.ast.tree.Relation
20+
import org.opensearch.sql.ast.tree.UnresolvedPlan
21+
import org.opensearch.sql.ppl.antlr.PPLSyntaxParser
22+
import org.opensearch.sql.ppl.parser.AstBuilder
23+
import org.opensearch.tasks.Task
24+
import org.opensearch.transport.TransportService
25+
26+
private val log = LogManager.getLogger(TransportExtractPPLIndicesAction::class.java)
27+
28+
class TransportExtractPPLIndicesAction @Inject constructor(
29+
transportService: TransportService,
30+
actionFilters: ActionFilters
31+
) : HandledTransportAction<ExtractPPLIndicesRequest, ExtractPPLIndicesResponse>(
32+
ExtractPPLIndicesAction.NAME, transportService, actionFilters, ::ExtractPPLIndicesRequest
33+
) {
34+
override fun doExecute(
35+
task: Task,
36+
request: ExtractPPLIndicesRequest,
37+
listener: ActionListener<ExtractPPLIndicesResponse>
38+
) {
39+
try {
40+
val query = request.query
41+
log.info("PPL_INDEX_EXTRACT: Received query: $query")
42+
43+
// Step 1: Parse PPL string into ANTLR parse tree
44+
val parser = PPLSyntaxParser()
45+
val parseTree = parser.parse(query)
46+
log.info("PPL_INDEX_EXTRACT: Parse tree created")
47+
48+
// Step 2: Convert parse tree to UnresolvedPlan AST
49+
val astBuilder = AstBuilder(query)
50+
val plan: UnresolvedPlan = astBuilder.visit(parseTree)
51+
log.info("PPL_INDEX_EXTRACT: AST built, root type=${plan.javaClass.simpleName}")
52+
53+
// Step 3: Walk the AST to extract Relation nodes
54+
val indices = mutableListOf<String>()
55+
extractIndices(plan, indices)
56+
log.info("PPL_INDEX_EXTRACT: Extracted indices=$indices")
57+
58+
listener.onResponse(ExtractPPLIndicesResponse(indices))
59+
} catch (e: Exception) {
60+
log.error("PPL_INDEX_EXTRACT: Failed to extract indices", e)
61+
listener.onFailure(e)
62+
}
63+
}
64+
65+
private fun extractIndices(node: Node, indices: MutableList<String>) {
66+
node.accept(
67+
object : AbstractNodeVisitor<Void?, Void?>() {
68+
override fun visitRelation(
69+
relation: Relation,
70+
context: Void?
71+
): Void? {
72+
val names = relation.qualifiedNames
73+
log.info(
74+
"PPL_INDEX_EXTRACT: Found Relation" +
75+
" type=${relation.javaClass.simpleName}" +
76+
" names=$names"
77+
)
78+
for (name: QualifiedName in names) {
79+
val raw = name.parts.joinToString(".")
80+
val cleaned = raw
81+
.replace(".PPL_MAPPINGS_ODFE_SYS_TABLE", "")
82+
.replace(".INFORMATION_SCHEMA_COLUMNS", "")
83+
val split = cleaned.split(",")
84+
.map { it.trim() }
85+
.filter { it.isNotEmpty() }
86+
indices.addAll(split)
87+
log.info(
88+
"PPL_INDEX_EXTRACT: Extracted=$split" +
89+
" from raw=$raw"
90+
)
91+
}
92+
return null
93+
}
94+
95+
override fun visitChildren(
96+
node: Node,
97+
context: Void?
98+
): Void? {
99+
for (child in node.child) {
100+
child.accept(this, context)
101+
}
102+
return null
103+
}
104+
},
105+
null
106+
)
107+
}
108+
}

‎core/build.gradle‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ dependencies {
4141
implementation 'commons-validator:commons-validator:1.7'
4242
implementation 'org.json:json:20240303'
4343

44-
api fileTree(dir: sqlJarDirectory, include: ["opensearch-sql-thin-${opensearch_build}.jar", "ppl-${opensearch_build}.jar", "protocol-${opensearch_build}.jar", "core-${opensearch_build}.jar"])
44+
api fileTree(dir: sqlJarDirectory, include: ["opensearch-sql-thin-${opensearch_build}.jar", "ppl-${opensearch_build}.jar", "protocol-${opensearch_build}.jar", "core-${opensearch_build}.jar", "common-${opensearch_build}.jar", "antlr4-runtime-*.jar", "guava-*.jar"])
4545

4646
zipArchive group: 'org.opensearch.plugin', name:'opensearch-sql-plugin', version: "${opensearch_build}"
4747

‎org/opensearch/remote/metadata/client/impl/LocalClusterIndicesClient.class‎

32.1 KB
Binary file not shown.

0 commit comments

Comments
 (0)