-
Notifications
You must be signed in to change notification settings - Fork 33
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Lukasz Soszynski <[email protected]>
- Loading branch information
1 parent
7649f9b
commit 735eb61
Showing
10 changed files
with
512 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
109 changes: 109 additions & 0 deletions
109
...t/src/integration/scala/org/opensearch/flint/spark/ppl/FlintSparkPPLFillnullITSuite.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package org.opensearch.flint.spark.ppl | ||
|
|
||
| import org.apache.spark.sql.{QueryTest, Row} | ||
| import org.apache.spark.sql.streaming.StreamTest | ||
|
|
||
| class FlintSparkPPLFillnullITSuite | ||
| extends QueryTest | ||
| with LogicalPlanTestUtils | ||
| with FlintPPLSuite | ||
| with StreamTest { | ||
|
|
||
| private val testTable = "spark_catalog.default.flint_ppl_test" | ||
|
|
||
| override def beforeAll(): Unit = { | ||
| super.beforeAll() | ||
|
|
||
| // Create test table | ||
| createNullableTableHttpLog(testTable) | ||
| } | ||
|
|
||
| protected override def afterEach(): Unit = { | ||
| super.afterEach() | ||
| // Stop all streaming jobs if any | ||
| spark.streams.active.foreach { job => | ||
| job.stop() | ||
| job.awaitTermination() | ||
| } | ||
| } | ||
|
|
||
| test("test fillnull with one null replacement value and one column") { | ||
| val frame = sql(s""" | ||
| | source = $testTable | fillnull value = 0 status_code | ||
| | """.stripMargin) | ||
|
|
||
| val results: Array[Row] = frame.collect() | ||
| val expectedResults: Array[Row] = | ||
| Array( | ||
| Row(1, "/home", null, 200), | ||
| Row(2, "/about", "2023-10-01 10:05:00", 0), | ||
| Row(3, "/contact", "2023-10-01 10:10:00", 0), | ||
| Row(4, null, "2023-10-01 10:15:00", 301), | ||
| Row(5, null, "2023-10-01 10:20:00", 200), | ||
| Row(6, "/home", null, 403)) | ||
| // Compare the results | ||
| implicit val rowOrdering: Ordering[Row] = Ordering.by[Row, Int](_.getAs[Int](0)) | ||
| assert(results.sorted.sameElements(expectedResults.sorted)) | ||
| } | ||
|
|
||
| test("test fillnull with various null replacement values and one column") { | ||
| val frame = sql(s""" | ||
| | source = $testTable | fillnull fields status_code=101 | ||
| | """.stripMargin) | ||
|
|
||
| val results: Array[Row] = frame.collect() | ||
| val expectedResults: Array[Row] = | ||
| Array( | ||
| Row(1, "/home", null, 200), | ||
| Row(2, "/about", "2023-10-01 10:05:00", 101), | ||
| Row(3, "/contact", "2023-10-01 10:10:00", 101), | ||
| Row(4, null, "2023-10-01 10:15:00", 301), | ||
| Row(5, null, "2023-10-01 10:20:00", 200), | ||
| Row(6, "/home", null, 403)) | ||
| // Compare the results | ||
| implicit val rowOrdering: Ordering[Row] = Ordering.by[Row, Int](_.getAs[Int](0)) | ||
| assert(results.sorted.sameElements(expectedResults.sorted)) | ||
| } | ||
|
|
||
| test("test fillnull with one null replacement value and two columns") { | ||
| val frame = sql(s""" | ||
| | source = $testTable | fillnull value = '???' request_path, timestamp | fields id, request_path, timestamp | ||
| | """.stripMargin) | ||
|
|
||
| val results: Array[Row] = frame.collect() | ||
| val expectedResults: Array[Row] = | ||
| Array( | ||
| Row(1, "/home", "???"), | ||
| Row(2, "/about", "2023-10-01 10:05:00"), | ||
| Row(3, "/contact", "2023-10-01 10:10:00"), | ||
| Row(4, "???", "2023-10-01 10:15:00"), | ||
| Row(5, "???", "2023-10-01 10:20:00"), | ||
| Row(6, "/home", "???")) | ||
| // Compare the results | ||
| implicit val rowOrdering: Ordering[Row] = Ordering.by[Row, Int](_.getAs[Int](0)) | ||
| assert(results.sorted.sameElements(expectedResults.sorted)) | ||
| } | ||
|
|
||
| test("test fillnull with various null replacement values and two columns") { | ||
| val frame = sql(s""" | ||
| | source = $testTable | fillnull fields request_path='/not_found', timestamp='*' | fields id, request_path, timestamp | ||
| | """.stripMargin) | ||
|
|
||
| val results: Array[Row] = frame.collect() | ||
| val expectedResults: Array[Row] = | ||
| Array( | ||
| Row(1, "/home", "*"), | ||
| Row(2, "/about", "2023-10-01 10:05:00"), | ||
| Row(3, "/contact", "2023-10-01 10:10:00"), | ||
| Row(4, "/not_found", "2023-10-01 10:15:00"), | ||
| Row(5, "/not_found", "2023-10-01 10:20:00"), | ||
| Row(6, "/home", "*")) | ||
| // Compare the results | ||
| implicit val rowOrdering: Ordering[Row] = Ordering.by[Row, Int](_.getAs[Int](0)) | ||
| assert(results.sorted.sameElements(expectedResults.sorted)) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
99 changes: 99 additions & 0 deletions
99
ppl-spark-integration/src/main/java/org/opensearch/sql/ast/tree/FillNull.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| package org.opensearch.sql.ast.tree; | ||
|
|
||
| import org.opensearch.sql.ast.AbstractNodeVisitor; | ||
| import org.opensearch.sql.ast.Node; | ||
| import org.opensearch.sql.ast.expression.Field; | ||
| import org.opensearch.sql.ast.expression.Literal; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Objects; | ||
| import java.util.stream.Collectors; | ||
| import java.util.stream.Stream; | ||
|
|
||
| public class FillNull extends UnresolvedPlan { ; | ||
|
|
||
| public static class NullableFieldFill { | ||
| private final Field nullableFieldReference; | ||
| private final Literal replaceNullWithMe; | ||
|
|
||
| public NullableFieldFill(Field nullableFieldReference, Literal replaceNullWithMe) { | ||
| this.nullableFieldReference = Objects.requireNonNull(nullableFieldReference, "Field to replace is required"); | ||
| this.replaceNullWithMe = Objects.requireNonNull(replaceNullWithMe, "Null replacement is required"); | ||
| } | ||
|
|
||
| public Field getNullableFieldReference() { | ||
| return nullableFieldReference; | ||
| } | ||
|
|
||
| public Literal getReplaceNullWithMe() { | ||
| return replaceNullWithMe; | ||
| } | ||
| } | ||
|
|
||
| private interface ContainNullableFieldFill { | ||
| Stream<NullableFieldFill> getNullFieldFill(); | ||
| } | ||
|
|
||
| public static class SameValueNullFill implements ContainNullableFieldFill { | ||
| private final List<NullableFieldFill> replacements; | ||
|
|
||
| public SameValueNullFill(Literal replaceNullWithMe, List<Field> nullableFieldReferences) { | ||
| Objects.requireNonNull(replaceNullWithMe, "Null replacement is required"); | ||
| this.replacements = Objects.requireNonNull(nullableFieldReferences, "Nullable field reference is required") | ||
| .stream() | ||
| .map(nullableReference -> new NullableFieldFill(nullableReference, replaceNullWithMe)) | ||
| .collect(Collectors.toList()); | ||
| } | ||
|
|
||
| @Override | ||
| public Stream<NullableFieldFill> getNullFieldFill() { | ||
| return replacements.stream(); | ||
| } | ||
| } | ||
|
|
||
| public static class VariousValueNullFill implements ContainNullableFieldFill { | ||
| private final List<NullableFieldFill> replacements; | ||
|
|
||
| public VariousValueNullFill(List<NullableFieldFill> replacements) { | ||
| this.replacements = replacements; | ||
| } | ||
|
|
||
| @Override | ||
| public Stream<NullableFieldFill> getNullFieldFill() { | ||
| return replacements.stream(); | ||
| } | ||
| } | ||
|
|
||
| private UnresolvedPlan child; | ||
| private final SameValueNullFill sameValueNullFill; | ||
| private final VariousValueNullFill variousValueNullFill; | ||
|
|
||
| public FillNull(SameValueNullFill sameValueNullFill, VariousValueNullFill variousValueNullFill) { | ||
| this.sameValueNullFill = sameValueNullFill; | ||
| this.variousValueNullFill = variousValueNullFill; | ||
| } | ||
|
|
||
| public List<NullableFieldFill> getNullableFieldFills() { | ||
| return Stream.of(sameValueNullFill, variousValueNullFill) | ||
| .filter(Objects::nonNull) | ||
| .flatMap(ContainNullableFieldFill::getNullFieldFill) | ||
| .collect(Collectors.toList()); | ||
| } | ||
|
|
||
| @Override | ||
| public UnresolvedPlan attach(UnresolvedPlan child) { | ||
| this.child = child; | ||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public List<? extends Node> getChild() { | ||
|
|
||
| return child == null ? List.of() : List.of(child); | ||
| } | ||
|
|
||
| @Override | ||
| public <T, C> T accept(AbstractNodeVisitor<T, C> nodeVisitor, C context) { | ||
| return nodeVisitor.visitFillNull(this, context); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.